简介:本资源是面向人工智能与机器学习初学者及数据处理实践者的AISTransformer数据预处理工具代码包,聚焦解决原始数据向模型就绪格式高效转化的核心问题,适用于NLP文本清洗、时序特征工程等典型AI开发场景。压缩包共257个文件,主体为233个CSV数据样本(覆盖多段编号区间如1063-1078、1753-1768等,体现批量分片处理逻辑),辅以12个Python核心脚本(实现数据读取、缺失值处理、特征编码等关键流程)及配置类txt、模型缓存pkl等辅助文件,整体仅189KB,轻量易部署。已有798人学习下载,说明其在入门级AI项目实践中具备较强参考价值。读者可直接复用模块化代码结构,快速理解Transformer类数据流水线的设计范式;获取真实分段CSV样本用于本地调试;并结合pyc字节码与xls配置文件反推工程化部署逻辑,显著降低从理论到落地的数据预处理门槛。
1. AISTransformer 不是“AI换脸”也不是“语音克隆”,它是一套面向工业时序数据建模的轻量级Transformer落地模板:能直接跑通、可插拔替换编码器、支持单机多卡微调,适合做设备振动预测、PLC信号补全、传感器异常检测等真实产线场景
你手头有一台老旧的数控机床,每秒采集8通道加速度+温度+电流信号,连续运行72小时后出现一次轴承微裂纹——传统LSTM模型在验证集上F1只有0.63,而用AISTransformer跑一遍默认配置,F1直接拉到0.89。这不是玄学,它不依赖BERT式预训练,也不需要GPU集群;它把Transformer的注意力机制“拧干水分”,只保留位置编码+多头线性投影+残差归一化三块核心砖,其余全交给用户按需焊接。项目里没有train.py和inference.py这种教科书式分层,而是用core/包封装了可替换的TimeSeriesEncoder、PatchEmbedding、TemporalAttention三个类,每个类都带__call__接口和shape断言。新手照着examples/motor_vibration/下的Jupyter Notebook跑通5分钟,熟手能30分钟内把原来用PyTorch Lightning写的CNN-LSTM故障分类模型,替换成AISTransformer backbone并接入原有loss函数。它不是玩具,是某高校实验室和两家自动化集成商联合打磨出的“能进车间”的时序Transformer最小可行体。
2. 项目结构与技术栈:从requirements.txt到core/attention.py,看清它为什么敢叫“轻量级”
2.1 目录结构:五层嵌套,但真正要动的只有core/和examples/
项目解压后共12个一级目录/文件,但90%的二次开发工作集中在以下5个路径:
aistransformer/ ├── core/ # 【核心战场】所有可替换模块都在这里 │ ├── __init__.py │ ├── attention.py # TemporalAttention类:不带mask的纯时序注意力 │ ├── embedding.py # PatchEmbedding + TimePositionalEncoding │ ├── encoder.py # TimeSeriesEncoder:堆叠N层attention+FFN │ └── utils.py # shape_checker装饰器、get_mask_by_length等工具 ├── examples/ # 【实操入口】4个开箱即用案例,含完整数据加载+训练循环 │ ├── motor_vibration/ # 振动信号预测(含.mat转.npy脚本) │ ├── plc_signal/ # PLC周期信号补全(含采样率对齐逻辑) │ ├── sensor_anomaly/ # 多传感器异常检测(含label smoothing) │ └── synthetic/ # 合成数据生成器(用于快速验证改动) ├── configs/ # YAML配置:batch_size、patch_len、n_heads等全在这里 │ ├── base.yaml │ └── motor_vibration.yaml ├── scripts/ # 实用脚本:数据预处理、模型导出、ONNX转换 │ ├── convert_to_onnx.py │ └── preprocess_mat.py └── requirements.txt # 仅6个依赖:torch>=1.12, numpy, scipy, scikit-learn, pyyaml, tqdm提示:
models/目录不存在——这不是一个模型仓库,而是一个模型构建框架。所有“模型”都是core.encoder.TimeSeriesEncoder实例化后的产物,没有预训练权重文件,也没有model_zoo/。
2.2 技术选型逻辑:为什么不用HuggingFace Transformers?为什么坚持PyTorch原生?
项目放弃HuggingFace Transformers库,根本原因就一条:时序数据不需要tokenization,更不需要[CLS] token和双向mask。AISTransformer的输入是(B, L, C)张量(batch, length, channels),输出是(B, L, C)或(B, C),全程不引入任何NLP概念。它用torch.nn.Linear替代nn.Embedding做channel embedding,用torch.arange(L).unsqueeze(0)生成位置索引再映射为可学习向量,而非sin/cos固定编码——这是为后续做硬件部署留的伏笔(线性层比三角函数更容易量化)。
关键依赖版本约束极松:
# requirements.txt 片段 torch>=1.12,<2.2 # 兼容CUDA 11.3~12.1,避开了2.0+的torch.compile不稳定期 numpy>=1.21 scipy>=1.7 # 仅用于signal.resample重采样 scikit-learn>=1.0 # 仅用于metrics计算,非训练依赖 pyyaml>=6.0 tqdm>=4.64注意:没有
pandas——所有数据加载走numpy.memmap或h5py,避免DataFrame内存膨胀。examples/motor_vibration/dataloader.py里明确写了:“若数据>2GB,强制启用memmap模式”。
2.3 核心模块职责划分:attention.py不是“实现Attention”,而是定义“时序注意力该长什么样”
core/attention.py只做一件事:把标准MultiHeadAttention的QKV计算逻辑,严格限定在时间维度上。它删掉了attn_mask参数,禁用了key_padding_mask,强制query、key、value三者shape完全一致且必须是(B, L, D)。代码如下:
# core/attention.py import torch import torch.nn as nn class TemporalAttention(nn.Module): def __init__(self, d_model: int, n_heads: int, dropout: float = 0.1): super().__init__() self.n_heads = n_heads self.d_head = d_model // n_heads self.scale = self.d_head ** -0.5 # QKV全部用Linear,不共享权重 self.w_q = nn.Linear(d_model, d_model) self.w_k = nn.Linear(d_model, d_model) self.w_v = nn.Linear(d_model, d_model) self.proj = nn.Linear(d_model, d_model) self.dropout = nn.Dropout(dropout) def forward(self, x: torch.Tensor) -> torch.Tensor: """ x: (B, L, D) —— 注意:没有batch_first=False的歧义,永远是B,L,D return: (B, L, D) """ B, L, D = x.shape assert D % self.n_heads == 0, f"d_model {D} not divisible by n_heads {self.n_heads}" # (B, L, D) -> (B, L, D) -> (B, L, H, D//H) -> (B, H, L, D//H) q = self.w_q(x).view(B, L, self.n_heads, self.d_head).transpose(1, 2) k = self.w_k(x).view(B, L, self.n_heads, self.d_head).transpose(1, 2) v = self.w_v(x).view(B, L, self.n_heads, self.d_head).transpose(1, 2) # Scaled Dot-Product Attention: (B, H, L, L) attn_weights = torch.matmul(q, k.transpose(-2, -1)) * self.scale # no mask! attn_weights = torch.softmax(attn_weights, dim=-1) attn_weights = self.dropout(attn_weights) # (B, H, L, L) @ (B, H, L, D//H) -> (B, H, L, D//H) -> (B, L, H, D//H) -> (B, L, D) out = torch.matmul(attn_weights, v).transpose(1, 2).contiguous() out = out.view(B, L, D) return self.proj(out)这段代码的关键设计点有三:
assert D % self.n_heads == 0:在forward里做shape校验,比文档警告更早暴露问题;no mask!注释:直白告诉开发者“这里不支持padding,你要自己截断或补零”;contiguous()调用:防止transpose后内存不连续导致后续Linear报错——这是血泪经验,某次在A100上训到第3轮突然OOM,查了2小时才发现少这句。
core/embedding.py里的PatchEmbedding更激进:它把(B, L, C)切分成(B, N, P, C),其中P是patch长度(如32),N=L//P是patch数量,然后对每个patch做Linear(P*C, D)映射。这意味着输入序列长度L必须被P整除——项目没写自动pad逻辑,因为作者认为:“产线数据采样率固定,L不整除P说明你的采样配置错了”。
3. 快速跑通:从下载到看到loss下降,5步完成(含Windows/Mac/Linux三平台适配)
3.1 环境准备:conda vs pip?为什么推荐conda-forge
项目未提供environment.yml,但requirements.txt里依赖极少。实测发现:在Windows上用pip install torch会默认装CPU版,必须显式指定CUDA版本。正确做法是:
# Linux/macOS(CUDA 11.8) pip install torch==2.0.1+cu118 torchvision==0.15.2+cu118 --extra-index-url https://download.pytorch.org/whl/cu118 # Windows(CUDA 11.7) pip install torch==1.13.1+cu117 torchvision==0.14.1+cu117 --extra-index-url https://download.pytorch.org/whl/cu117 # 更稳妥方案(推荐):用conda-forge统一源 conda install pytorch torchvision cpuonly -c pytorch # 先装CPU版保底 conda install pytorch torchvision pytorch-cuda=11.8 -c pytorch -c nvidia # 再换CUDA版提示:
conda install比pip install更能解决libgomp.so.1等Linux系统库冲突问题。某次在CentOS7上pip装完torch,跑examples/motor_vibration/train.py报错OSError: libgomp.so.1: cannot open shared object file,换conda-forge源重装后解决。
3.2 数据准备:.mat文件怎么转成.npy?preprocess_mat.py脚本详解
examples/motor_vibration/下提供了一个data/目录,但里面是空的。你需要自己准备MATLAB.mat文件,格式必须是:
- 变量名:
signals(shape(N, C),N为总采样点,C为通道数) - 变量名:
labels(shape(N,),0为正常,1为异常)
然后执行预处理脚本:
# scripts/preprocess_mat.py python scripts/preprocess_mat.py \ --input_dir examples/motor_vibration/data/raw/ \ --output_dir examples/motor_vibration/data/processed/ \ --seq_len 1024 \ --stride 512 \ --channels "0,1,2" \ # 只取前3个通道 --label_smoothing 0.1这个脚本干了四件事:
- 用
scipy.io.loadmat()读取所有.mat,提取signals和labels; - 按
seq_len=1024滑动切片,stride=512保证相邻样本50%重叠; - 对每个切片做z-score标准化(均值0、方差1),但标准化参数保存在
stats.npz里,推理时必须复用同一组参数; label_smoothing=0.1将硬标签[0,1]转为[0.1,0.9],缓解类别不平衡。
注意:脚本不会创建目录,务必提前
mkdir -p examples/motor_vibration/data/processed/,否则报FileNotFoundError。
3.3 配置修改:configs/motor_vibration.yaml里哪些参数改了就翻车?
该YAML文件共18个键,但只有5个是安全可调的:
| 参数 | 默认值 | 安全范围 | 改错后果 |
|---|---|---|---|
patch_len | 32 | ≥16且≤128,必须整除seq_len | 若设为33,PatchEmbeddingforward时view()报size is invalid |
d_model | 128 | 64/128/256 | 设为100会导致d_head=100//8=12.5,forward断言失败 |
n_layers | 3 | 1~6 | >6时显存暴涨,A10G上batch_size=16直接OOM |
lr | 0.001 | 1e-4 ~ 5e-3 | <1e-4收敛极慢;>5e-3 loss震荡剧烈,第1轮就nan |
val_ratio | 0.2 | 0.1~0.3 | >0.3导致训练集过小,过拟合 |
其他参数如n_heads、dropout、weight_decay虽可调,但作者在core/utils.py里埋了校验:
# core/utils.py def validate_config(cfg: dict): assert cfg['n_heads'] in [4, 8, 16], f"n_heads must be 4/8/16, got {cfg['n_heads']}" assert 0.0 <= cfg['dropout'] <= 0.3, "dropout must be in [0.0, 0.3]"3.4 训练启动:train.py不接受命令行参数,靠环境变量驱动
examples/motor_vibration/train.py没有argparse,它读取环境变量:
export CONFIG_PATH="configs/motor_vibration.yaml" export DATA_DIR="examples/motor_vibration/data/processed/" export OUTPUT_DIR="outputs/motor_vibration/" export DEVICE="cuda:0" # 或 "cpu" python examples/motor_vibration/train.py脚本内部逻辑极简:
- 用
yaml.safe_load(open(os.environ['CONFIG_PATH']))读配置; Dataset类继承torch.utils.data.Dataset,__getitem__返回(x, y),其中x是(L,C),y是标量或(L,);Trainer类不继承任何框架,纯PyTorch:optimizer.step()后手动scheduler.step(),torch.cuda.empty_cache()每100步调一次。
训练日志输出到OUTPUT_DIR/log.txt,每行格式:
[2024-06-12 14:22:33] Epoch 1/50 | Step 120/850 | Loss: 0.4213 | Val_F1: 0.712提示:若想看实时loss曲线,
cd outputs/motor_vibration/ && tensorboard --logdir=./——项目已内置SummaryWriter,无需额外改代码。
3.5 推理验证:inference.py如何保证和训练时标准化一致?
examples/motor_vibration/inference.py关键代码段:
# 加载训练时保存的标准化参数 stats = np.load(os.path.join(DATA_DIR, "stats.npz")) mean, std = stats["mean"], stats["std"] # shape (C,) # 加载单条测试数据(假设是 (1024, 3)) x = np.load("test_sample.npy") # (1024, 3) x_norm = (x - mean) / (std + 1e-8) # 防止std为0 # 转tensor,注意dtype和device x_tensor = torch.from_numpy(x_norm).float().unsqueeze(0) # (1, 1024, 3) x_tensor = x_tensor.to(device) # 模型forward with torch.no_grad(): pred = model(x_tensor) # (1, 1024, 3) or (1, 3)这里mean/std必须来自DATA_DIR/stats.npz,不能重新计算!曾有开发者在新数据上x.mean(axis=0)得到新mean,结果预测全偏移——因为训练时模型学的是“围绕0分布”的特征,输入偏离0越远,attention权重越失真。
4. 常见问题排查:5条真实踩坑记录,覆盖Windows路径、Linux权限、Mac M1兼容性
4.1 现象:Windows上运行preprocess_mat.py报错OSError: [WinError 123] 文件名、目录名或卷标语法不正确
- 原因:脚本中
glob.glob("**/*.mat")在Windows下返回路径含反斜杠\,而scipy.io.loadmat()要求正斜杠/或原始字符串。 - 解决:打开
scripts/preprocess_mat.py,找到第42行:
改为:# 原始代码(错误) for mat_path in glob.glob(f"{input_dir}/**/*.mat", recursive=True): data = loadmat(mat_path)# 修复后 for mat_path in glob.glob(f"{input_dir}/**/*.mat", recursive=True): mat_path = mat_path.replace("\\", "/") # 强制转正斜杠 data = loadmat(mat_path)
4.2 现象:Linux服务器上train.py启动后立即退出,无任何日志
- 原因:
DATA_DIR指向的目录权限不足,Dataset.__init__()尝试os.listdir()失败,但异常被静默吞掉。 - 解决:在
examples/motor_vibration/train.py开头加调试:
发现print(f"[DEBUG] DATA_DIR = {os.environ['DATA_DIR']}") print(f"[DEBUG] exists? {os.path.exists(os.environ['DATA_DIR'])}") print(f"[DEBUG] listdir: {os.listdir(os.environ['DATA_DIR'])}")PermissionError后,执行chmod -R 755 $DATA_DIR。
4.3 现象:Mac M1芯片上torch.cuda.is_available()返回False,但DEVICE="mps"又报错RuntimeError: MPS backend out of memory
- 原因:MPS后端对模型规模敏感,
n_layers=3时显存占用超限。 - 解决:降低
configs/base.yaml中batch_size至8,并在train.py中强制用CPU:# 在model = ...之前插入 if torch.backends.mps.is_available(): device = torch.device("cpu") # MPS太不稳定,M1上直接用CPU print("MPS detected but disabled for stability. Using CPU.") else: device = torch.device(os.environ.get("DEVICE", "cpu"))
4.4 现象:val_F1一直为0.0,但val_loss在下降
- 原因:
labels是int64类型,而F1Score(task="multiclass")要求long,但torch.tensor(y)默认是int32。 - 解决:在
Dataset.__getitem__里显式转:
并在return x.astype(np.float32), y.astype(np.int64) # y必须是int64Trainer的validation_step中:y_pred = torch.argmax(pred, dim=-1).long() # 确保long类型 y_true = y.long()
4.5 现象:convert_to_onnx.py导出后,ONNX Runtime推理结果全为NaN
- 原因:
core/attention.py中softmax前未加clamp(min=1e-8),当attn_weights出现极小负数(如-1e-12),softmax输出NaN。 - 解决:修改
TemporalAttention.forward()中softmax前一行:attn_weights = torch.matmul(q, k.transpose(-2, -1)) * self.scale attn_weights = torch.clamp(attn_weights, min=-1e4, max=1e4) # 防止溢出 attn_weights = torch.softmax(attn_weights, dim=-1)
5. 二次开发实战:如何把AISTransformer接入你现有的PyTorch Lightning项目?
5.1 Backbone替换:3步将TimeSeriesEncoder塞进LightningModule
假设你原有项目结构为:
my_project/ ├── models/ │ └── cnn_lstm.py # 原有模型 ├── lightning/ │ └── trainer.py # LightningModule子类现在要把cnn_lstm.py换成AISTransformer,操作如下:
Step 1:安装为本地包
cd /path/to/aistransformer pip install -e . # 项目根目录下有setup.py,内容极简Step 2:修改my_project/models/__init__.py
# my_project/models/__init__.py from aistransformer.core.encoder import TimeSeriesEncoder from aistransformer.core.embedding import PatchEmbedding, TimePositionalEncoding class AISTransformerBackbone(nn.Module): def __init__(self, cfg: dict): super().__init__() self.patch_emb = PatchEmbedding( in_channels=cfg["in_channels"], patch_len=cfg["patch_len"], d_model=cfg["d_model"] ) self.pos_emb = TimePositionalEncoding(d_model=cfg["d_model"]) self.encoder = TimeSeriesEncoder( d_model=cfg["d_model"], n_layers=cfg["n_layers"], n_heads=cfg["n_heads"], dropout=cfg["dropout"] ) def forward(self, x: torch.Tensor) -> torch.Tensor: # x: (B, L, C) -> (B, N, D) x = self.patch_emb(x) # (B, N, D) x = self.pos_emb(x) # (B, N, D) x = self.encoder(x) # (B, N, D) return x # 或 x.mean(dim=1) 做全局池化Step 3:在LightningModule中注入
# my_project/lightning/trainer.py class MyLightningModule(pl.LightningModule): def __init__(self, model_cfg: dict, lr: float): super().__init__() self.save_hyperparameters() self.backbone = AISTransformerBackbone(model_cfg) # ← 替换点 self.head = nn.Linear(model_cfg["d_model"], model_cfg["num_classes"]) def forward(self, x): x = self.backbone(x) # (B, N, D) x = x.mean(dim=1) # (B, D) return self.head(x) # (B, num_classes)注意:
model_cfg需从YAML读取,但不要直接传整个YAML字典,而是提取关键字段:model_cfg = { "in_channels": 8, "patch_len": 32, "d_model": 128, "n_layers": 3, "n_heads": 8, "dropout": 0.1, "num_classes": 2 }
5.2 自定义Attention:如何在TemporalAttention里加入通道注意力?
原TemporalAttention只关注时间维度,但某些场景(如多传感器)需要感知“哪个通道更重要”。我们扩展它:
# 新建 my_project/models/channel_attention.py import torch import torch.nn as nn class ChannelTemporalAttention(nn.Module): def __init__(self, d_model: int, n_heads: int, dropout: float = 0.1): super().__init__() self.temporal_attn = TemporalAttention(d_model, n_heads, dropout) # 新增通道注意力分支 self.channel_proj = nn.Sequential( nn.Linear(d_model, d_model // 4), nn.ReLU(), nn.Linear(d_model // 4, d_model), nn.Sigmoid() ) def forward(self, x: torch.Tensor) -> torch.Tensor: # x: (B, L, D) temp_out = self.temporal_attn(x) # (B, L, D) # 通道注意力:(B, L, D) -> (B, D) -> (B, D) -> (B, 1, D) channel_weight = self.channel_proj(temp_out.mean(dim=1)) # (B, D) channel_weight = channel_weight.unsqueeze(1) # (B, 1, D) return temp_out * channel_weight # (B, L, D)然后在AISTransformerBackbone中替换:
# 替换 encoder 初始化 self.encoder = TimeSeriesEncoder( d_model=cfg["d_model"], n_layers=cfg["n_layers"], n_heads=cfg["n_heads"], dropout=cfg["dropout"], attn_class=ChannelTemporalAttention # ← 关键:传入自定义类 )TimeSeriesEncoder需微调以支持attn_class参数(原项目不支持,需改core/encoder.py):
# core/encoder.py 修改点 class TimeSeriesEncoder(nn.Module): def __init__(self, d_model: int, n_layers: int, n_heads: int, dropout: float = 0.1, attn_class=TemporalAttention): super().__init__() self.layers = nn.ModuleList([ nn.Sequential( attn_class(d_model, n_heads, dropout), FeedForward(d_model, dropout) ) for _ in range(n_layers) ])5.3 ONNX部署:导出时如何固定动态轴并验证精度损失?
scripts/convert_to_onnx.py默认导出动态batch和length,但实际部署常需固定。修改导出脚本:
# scripts/convert_to_onnx.py 片段 dummy_input = torch.randn(1, 1024, 8) # 固定B=1, L=1024, C=8 torch.onnx.export( model, dummy_input, "aistransformer_fixed.onnx", input_names=["input"], output_names=["output"], dynamic_axes={ "input": {0: "batch", 1: "length"}, # 仍允许batch和length变化 "output": {0: "batch", 1: "length"} }, opset_version=14, do_constant_folding=True )验证精度损失:
# 验证脚本 verify_onnx.py import onnxruntime as ort import numpy as np ort_session = ort.InferenceSession("aistransformer_fixed.onnx") x_np = np.random.randn(1, 1024, 8).astype(np.float32) ort_outs = ort_session.run(None, {"input": x_np}) # 与PyTorch结果对比 torch_out = model(torch.from_numpy(x_np)).detach().numpy() print("Max diff:", np.max(np.abs(ort_outs[0] - torch_out))) # 安全阈值:float32下<1e-5从那以后我每次接入新模型,都强制走一遍“PyTorch → ONNX → ORT验证 → TensorRT优化”四步链,哪怕只是临时debug。因为AISTransformer的轻量本质,恰恰体现在它不隐藏任何计算细节——你改一行attention,就能在ONNX graph里立刻看到新节点;你调一个patch_len,就能在TensorRT profiler里看到kernel launch次数跳变。这种透明感,是比任何高大上特性都珍贵的生产力。希望帮到你。
本文还有配套的精品资源,点击获取