简介：AISTransformer 数据代码面向从事人工智能与机器学习数据处理的开发者与学习者，聚焦原始数据的清洗、特征工程与格式转换，可作为构建训练数据集的参考工具。压缩包共257个文件，以233个CSV数据分片和12个Python源码脚本为主，另有少量编译缓存、Excel表格、说明文本及序列化文件，整体约189KB，体积轻便，便于按模块查阅与复用。已有799人学习。源码部分覆盖模块划分、函数与类实现、配置项及示例测试等层面，配合CSV样本可帮助读者理解数据读取、缺失值处理、特征编码和数据集构建的流程，也可作为自然语言处理、图像识别等场景预处理思路的对照材料，适合需要快速搭建数据处理管线或拆解AI数据代码结构的中级学习者。

1. 拿到一份「数据 + 代码」的 Transformer 包，先不要急着安装

有人把一个 Transformer 动作序列建模的压缩包丢给我，说「数据代码都在里面，跑一下」。解压后只有三样东西：data/下预处理好的张量、models/下的网络定义、scripts/下的训练与推理入口，外加一个可以直接装进别的工程当插件用的pyproject.toml。这类「数据 + 代码」包最容易出问题的地方从来不是模型本身，而是数据格式对不上：帧率差一点、关节顺序颠倒、旋转表示不统一，模型照样能跑完整个 epoch，loss 曲线也好看，但生成结果是抖动噪声。这份资源适合两类人：一类是手上有自己的动作或时序数据，想找个现成的 Transformer 训练骨架改；另一类是想把这套推理能力当插件嵌进自己的软件里。往下读之前先记住一件事：跑通不难，跑对才有价值。

2. 工程骨架与技术选型：把一份 Transformer 训练代码拆成四层

2.1 目录结构：四个包各自的边界

这份代码的目录划分是比较克制的，没有把所有东西塞进一个train.py，而是按职责切成数据、模型、引擎、命令行四层。我把它整理成下面这张表，看一遍就知道改东西该动哪个文件：

数据层之所以单独成包，是因为这类项目最耗时的部分永远是特征对齐，不是网络结构。模型层只有两个文件：embedding.py放位置编码和模态投影，transformer.py放编码器。引擎层把所有跟torch优化器、混合精度、学习率调度相关的逻辑收在一起，命令行层只做参数解析和对象装配。这种切法的好处是：你想把模型接到自己的服务里时，只需要from dmmotion.models.transformer import MotionTransformer，不用把训练脚本一起拖进来，插件化的前提就是模块之间没有循环依赖。

有一点需要提前说明：下面提到的数据集是一个公开的三维舞蹈动作序列数据集，我在文中统一用DMotion代称，避免和具体出处绑定。代码结构本身与数据集无关，换成任何「时序特征 + 对齐标签」的数据都能跑。

2.2 技术栈与最小可跑环境

依赖不算重，核心就是 PyTorch 和 numpy，剩下的都是为了工程化：pyyaml读配置，tqdm看进度，tensorboard看曲线，einops做维度重排，可选的torch.cuda.amp走混合精度。版本上有一个硬约束：nn.TransformerEncoderLayer的batch_first参数在较早的版本里没有，norm_first也是后加的，所以 PyTorch 至少要用 1.11 以上，我实测 2.0 和 2.1 都正常。

# 建一个干净环境，Python 3.9 是比较稳的选择 conda create -n dmmotion python=3.9 -y conda activate dmmotion # 按显卡情况装对应 CUDA 版本的 torch，别直接 pip install torch 拉到 CPU 版 pip install torch==2.1.0 torchvision --index-url https://download.pytorch.org/whl/cu118 pip install numpy tqdm pyyaml tensorboard einops # 以可编辑模式装成本地包，改代码不用重装 pip install -e . # 确认入口注册成功，会打印子命令列表 dmm --help

最后一条命令是关键。很多包用pip install -e .装完之后import正常但命令行找不到，问题都出在pyproject.toml里没写[project.scripts]，或者写了但main函数的签名不对。装完先验证这一条，比后面调半天路径要省事得多。

2.3 配置驱动：一次训练由哪几个字段决定

这套代码把可变项全抽到 yaml 里了，最常改的就是下面这几个字段：

# configs/dmotion_small.yaml data: feat_dir: data/features # 特征根目录，下面按序列 id 分词存放 train_index: data/split/train.jsonl val_index: data/split/val.jsonl window: 120 # 单个样本帧数，30fps 下约 4 秒 stride: 60 # 滑窗步长，训练集用一半重叠做增广 model: d_model: 512 nhead: 8 num_layers: 4 dim_feedforward: 1024 dropout: 0.1 train: batch_size: 64 epochs: 200 lr: 2.0e-4 warmup_steps: 1000 weight_decay: 0.01 grad_clip: 1.0 amp: true

window和stride是最值得先动的两个数。窗口太短，模型看不到完整的动作周期，生成结果会「一顿一顿」；窗口太长，显存吃不消且位置编码要跟着放大。我的经验是先把窗口设成数据里一个完整动作片段长度的 1.5 倍左右，再在验证集上往下调。batch_size和lr是绑定的，把 batch 从 64 拉到 128 而 lr 不动，等价于把有效学习率砍半，训练会显得「怎么都不收敛」。warmup_steps别设成 0，Transformer 在训练初期对学习率非常敏感，前一千步线性爬坡能明显减少早期的 loss 尖峰。

3. 数据管线拆解：从原始动作序列到可训练的批次

3.1 动作特征怎么定：6D 旋转 + 根节点位移

模型对输入表示的敏感程度远超想象。如果直接用欧拉角表示关节旋转，在接近 ±90 度的地方会出现跳变，网络学到的是一堆断层；如果用四元数，又要处理双倍覆盖（q 和 -q 表示同一旋转）。所以这份代码用的是连续 6D 旋转表示：每个关节取旋转矩阵的前两列，展平成 6 个数，再用 Gram-Schmidt 正交化还原。单帧特征维度是 24 个关节 × 6 = 144，加上根节点的 3 维位移，总共 147 维。

# dmmotion/data/features.py import numpy as np FEAT_DIM = 147 # 24 关节 x 6D 旋转(144) + 根节点位移(3) FPS = 30 def rotmat_to_6d(rot: np.ndarray) -> np.ndarray: """(T, J, 3, 3) 旋转矩阵 -> (T, J*6)，取前两列保证连续性""" t, j = rot.shape[:2] return rot[..., :2].reshape(t, j * 6).astype(np.float32) def build_windows(motion: np.ndarray, stride: int, window: int): """把整段序列切成等长窗口，返回 [(start, end), ...]""" total = motion.shape[0] if total < window: return [] # 太短的序列直接丢掉，别 pad 成一条噪声样本 starts = list(range(0, total - window + 1, stride)) if starts[-1] + window < total: # 末尾不足一窗的补一个，避免丢掉最后半秒 starts.append(total - window) return [(s, s + window) for s in starts]

这里有两个容易忽略的点。第一，build_windows对末尾做了补窗处理，否则序列尾部的动作永远进不了训练集，推理时模型对「结尾」这种模式完全没见过。第二，短于一个窗口的序列是直接丢弃而不是 pad，因为补零之后模型会把这些零当成真实姿态去学，反而是负样本。如果你的数据里短序列占比很高，正确做法是减小window，而不是无脑 pad。

3.2 滑窗与索引：把长序列切成等长样本

索引文件用 jsonl 而不是直接存切好的张量，是个很实用的取舍：切好的样本会带来几十倍的存储膨胀，尤其当 stride 只有 window 的一半时。jsonl 每行只记seq、start、end，试错成本极低，改 stride 只需要重跑一遍索引构建，几秒钟的事。

# scripts/build_index.py 的核心逻辑 import json, os import numpy as np from dmmotion.data.features import build_windows def build(split_dir, feat_dir, out_path, window, stride): with open(out_path, "w") as f: for name in sorted(os.listdir(split_dir)): seq_id = name.replace(".npy", "") motion = np.load(os.path.join(feat_dir, f"{seq_id}.motion.npy"), mmap_mode="r") audio = np.load(os.path.join(feat_dir, f"{seq_id}.audio.npy"), mmap_mode="r") # 动作和音频必须等长，不等长说明预处理阶段对齐出了问题 assert motion.shape[0] == audio.shape[0], f"{seq_id} 音画帧数不一致" for s, e in build_windows(motion, stride, window): f.write(json.dumps({"seq": seq_id, "start": s, "end": e}) + "\n")

那句assert是血泪经验。动作特征和音频特征通常是两条流水线分别产出的，一边按视频帧采样、一边按音频 hop 采样，差个一两帧很常见。差一帧在 assert 处直接炸掉，比训练到第 30 个 epoch 才发现标签错位要划算得多。如果你接手的数据集没有这个检查，建议补上。

3.3 Dataset 与归一化统计量：别让验证集漏进均值

Dataset 本身没什么花活，用mmap_mode="r"读.npy是重点——特征文件动辄几个 G，全部读进内存会直接把机器打满，mmap 让操作系统按需换页，十几个 G 的数据也能在 32G 内存的机器上跑起来。

# dmmotion/data/dataset.py import json import numpy as np import torch class MotionWindowDataset(torch.utils.data.Dataset): def __init__(self, index_path, feat_dir, mean, std): self.items = [json.loads(line) for line in open(index_path)] self.feat_dir = feat_dir self.mean = mean.reshape(1, -1).astype(np.float32) self.std = std.reshape(1, -1).astype(np.float32) def __len__(self): return len(self.items) def __getitem__(self, i): it = self.items[i] s, e = it["start"], it["end"] m = np.load(f"{self.feat_dir}/{it['seq']}.motion.npy", mmap_mode="r")[s:e] a = np.load(f"{self.feat_dir}/{it['seq']}.audio.npy", mmap_mode="r")[s:e] m = (np.asarray(m, dtype=np.float32) - self.mean) / self.std return {"motion": torch.from_numpy(m), "audio": torch.from_numpy(np.asarray(a, dtype=np.float32))} def collate(batch): # 窗口等长，直接 stack；若改成变长，这里要 pad 并返回 key_padding_mask return {k: torch.stack([b[k] for b in batch]) for k in ("motion", "audio")}

mean和std必须只在训练集上统计，这是一条不能破的线。我见过有实现图省事，把整个数据集的均值方差算一遍存成.npz，结果验证集的信息通过归一化泄漏进训练，指标好看得离谱，上线之后原形毕露。统计量建议用分块累加的方式算，避免把几百个 G 的数据一次性堆进内存：

def compute_stats(index_path, feat_dir, dim): n, s1, s2 = 0, np.zeros(dim), np.zeros(dim) for line in open(index_path): it = json.loads(line) x = np.load(f"{feat_dir}/{it['seq']}.motion.npy", mmap_mode="r")[it["start"]:it["end"]] n += x.shape[0]; s1 += x.sum(0); s2 += (x ** 2).sum(0) mean = s1 / n std = np.sqrt(np.maximum(s2 / n - mean ** 2, 1e-6)) # 夹一个下限，防止常量维度除零 return mean, std

最后那个1e-6的下限不是可选项。动作数据里有些维度长期不变（比如某些不动的末端关节），标准差算出来是 0，除下去整批特征直接变inf，训练第一步就 NaN。这种问题报错信息不会告诉你原因，只能靠提前防。

提示：换数据集之后一定要重新算一遍mean/std并存成新文件，不要复用旧数据集统计量。特征的物理含义变了，尺度的意义也就变了。

4. 模型与训练：Transformer 在这份代码里的具体实现

4.1 位置编码与掩码：序列模型的两个开关

Transformer 本身对顺序没有感知，位置编码不是「可选优化」，是必需品。这份代码用的是标准的正弦位置编码，实现时最容易错的地方是div_term的指数步长和pe的 buffer 注册——忘了persistent=False会让权重文件里多出一堆没用的张量，模型换个max_len就加载失败。

# dmmotion/models/embedding.py import math import torch import torch.nn as nn class SinusoidalPE(nn.Module): def __init__(self, d_model: int, max_len: int = 1024): super().__init__() pe = torch.zeros(max_len, d_model) pos = torch.arange(0, max_len).unsqueeze(1).float() div = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model)) pe[:, 0::2] = torch.sin(pos * div) pe[:, 1::2] = torch.cos(pos * div) # persistent=False：不写进 state_dict，避免换长度时加载报 missing keys self.register_buffer("pe", pe.unsqueeze(0), persistent=False) def forward(self, x): # x: (B, T, D) return x + self.pe[:, : x.size(1)]

掩码这块要说清楚：这份代码做的是「音频帧对齐动作帧」的回归任务，不是自回归解码，所以编码器里不加因果掩码，只在数据变长时加src_key_padding_mask（形状(B, T)，True 表示该位置是 padding）。这一点经常被误解，看到 Transformer 就以为一定要 causal mask，加上之后模型只能看历史帧，反而把同一时刻的音频信息挡在外面，指标掉一大截。如果你确实要做逐帧生成，才需要torch.triu生成的因果掩码，并且训练时用 teacher forcing。

4.2 编码器主体：音频与动作怎么融合

融合方式选的是最朴素的「早期相加」：音频特征和动作特征各自投影到d_model维，直接相加再进编码器。听起来简单，但对齐任务里它比 cross-attention 更稳，因为两个模态在时间轴上是一一对应的，不存在需要学对齐关系的必要。

# dmmotion/models/transformer.py import torch import torch.nn as nn from .embedding import SinusoidalPE from ..data.features import FEAT_DIM, AUDIO_DIM class MotionTransformer(nn.Module): def __init__(self, d_model=512, nhead=8, num_layers=4, dim_feedforward=1024, dropout=0.1): super().__init__() self.audio_proj = nn.Linear(AUDIO_DIM, d_model) self.motion_proj = nn.Linear(FEAT_DIM, d_model) self.pe = SinusoidalPE(d_model) layer = nn.TransformerEncoderLayer( d_model=d_model, nhead=nhead, dim_feedforward=dim_feedforward, dropout=dropout, activation="gelu", batch_first=True, norm_first=True, # pre-norm 更容易训深 ) self.encoder = nn.TransformerEncoder(layer, num_layers, enable_nested_tensor=False) self.norm = nn.LayerNorm(d_model) self.head = nn.Linear(d_model, FEAT_DIM) def forward(self, audio, motion_in, key_padding_mask=None): h = self.audio_proj(audio) + self.motion_proj(motion_in) h = self.pe(h) h = self.encoder(h, src_key_padding_mask=key_padding_mask) return self.head(self.norm(h)) # (B, T, FEAT_DIM)

几个参数值得单独说。norm_first=True走的是 pre-norm 结构，4 层以内差别不大，但如果你照着论文改成 8 层 12 层，post-norm 版本不加 warmup 基本训不动。enable_nested_tensor=False是为了避开某些 PyTorch 版本在带 mask 时的嵌套张量优化路径，那个路径在 2.0 早期版本上有过数值不一致的问题，关掉之后速度略降但结果可复现。activation="gelu"相比 ReLU 在回归任务上收敛更平滑，代价是前向慢一点点，可以接受。

注意motion_in这个输入。训练时它是「目标序列左移一帧」，推理时是模型自己的历史输出，两者分布不完全一致，这就是典型的 exposure bias。缓解办法后面在训练循环里用速度损失做一部分补偿，但别指望彻底消除，生成类任务都这样。

4.3 训练循环：AMP、梯度裁剪与速度损失

损失函数是两项相加：逐帧特征的 MSE，加上一阶差分（速度）的 MSE。只做逐帧 MSE 的话，模型倾向于输出「平均姿态」，看起来不抖但完全没有动作幅度；加上速度项之后，帧间的变化量被显式约束，动作才有「劲」。

# dmmotion/engine/train.py 的核心训练步 import torch from torch import nn mse = nn.MSELoss() def train_one_epoch(model, loader, opt, sched, scaler, device, clip=1.0, vel_w=0.5): model.train() total = 0.0 for batch in loader: audio = batch["audio"].to(device, non_blocking=True) target = batch["motion"].to(device, non_blocking=True) # 输入左移一帧：让模型学会"给定历史 + 当前音频，预测下一帧" motion_in = torch.cat([torch.zeros_like(target[:, :1]), target[:, :-1]], dim=1) with torch.cuda.amp.autocast(enabled=scaler is not None): pred = model(audio, motion_in) loss = mse(pred, target) loss = loss + vel_w * mse(pred[:, 1:] - pred[:, :-1], target[:, 1:] - target[:, :-1]) opt.zero_grad(set_to_none=True) scaler.scale(loss).backward() scaler.unscale_(opt) # 先反缩放再裁剪，顺序不能反 torch.nn.utils.clip_grad_norm_(model.parameters(), clip) scaler.step(opt) scaler.update() sched.step() total += loss.item() * audio.size(0) return total / len(loader.dataset)

scaler.unscale_(opt)必须在clip_grad_norm_之前调用，否则裁的是被放大了 65536 倍的梯度，等于没裁。这个顺序错了不会报错，只是训练偶尔崩一下，很难查。set_to_none=True省一点显存，在 24G 卡上跑batch_size=64差别不大，但在地 Reasoning 卡上是能多塞几个样本的。

启动训练：

# 单卡 dmm train --config configs/dmotion_small.yaml --tag run01 \ --resume auto # 有断点就续，没有就从头开始 # 只想快速验证管线通不通，把 epoch 压到 2、训练集裁到 200 条 dmm train --config configs/dmotion_small.yaml --debug --epochs 2

--resume auto会去runs/{tag}/last.ckpt找断点，同时恢复优化器和 scheduler 的状态。只存模型权重不存优化器状态是最常见的续训事故，恢复后 loss 会突然跳高一段，看起来像模型被reset了，其实就是 Adam 的动量项丢了。

5. 排查现场：跑不通的问题九成集中在这五处

5.1 环境与张量形状类

现象一：RuntimeError: expected scalar type Float but found Double。原因：从 numpy 读进来的数组默认是float64，torch.from_numpy会原样保留，跟模型参数的float32对不上。这在换了数据处理脚本之后特别容易出现，因为旧脚本里可能有一句.astype(np.float32)被顺手删了。 解决：在 Dataset 的__getitem__里显式指定 dtype，别依赖上游；再稳一点可以在collate之后统一.float()。我一般在 Dataset 出口处就卡死，不让这个问题流到训练循环里。

现象二：加载权重时报Missing key(s) in state_dict: ...pe，或者换了max_len之后加载失败。原因：位置编码的 buffer 被写进了state_dict，模型结构和权重文件里的形状对不上。另一种情况是用了DataParallel训练、单卡加载，权重 key 前面多了module.。 解决：register_buffer(..., persistent=False)让 PE 不入权重文件；多卡权重用{k.replace('module.', ''): v for k, v in sd.items()}过一遍再load_state_dict。

现象三：DataLoader卡在第一个 batch 不动，CPU 占满但 GPU 空转。原因：num_workers > 0时，每个 worker 都开了 mmap 去读同一批大文件，加上默认的prefetch_factor=2，在共享内存受限的容器里很容易互相等锁。 解决：先把num_workers设成 0 确认逻辑没问题，再逐步加到 4 或 8；同时把persistent_workers=True打开，避免每个 epoch 重建进程池。容器里还要检查/dev/shm大小，默认 64M 的话必须调大，否则 worker 直接被 kill。

5.2 数值与指标类

现象四：前几十步 loss 正常下降，然后突然变成nan。原因排下来无非三种：学习率太大加上没有 warmup；归一化时某个维度标准差为 0 导致除零；混合精度下梯度溢出但没触发缩放回退。 解决：先看是哪一步开始 NaN 的，如果是第 1 步，八成是除零，去查std里有没有 0；如果是几百步之后，把lr降一个数量级、加上warmup_steps，并把grad_clip从 1.0 收到 0.5。AMP 这边确认scaler.unscale_顺序没写反。

现象五：验证集 loss 比训练集还低，指标好看得不像话。原因：归一化统计量用了全量数据，验证集信息提前泄漏；或者验证时忘了model.eval()，dropout 还在随机丢特征，反而起到了集成效果，让指标虚高；再或者滑窗 stride 设得太小，训练集和验证集有大量重叠窗口。 解决：统计量只从训练索引算；验证循环开头老老实实model.eval()加torch.no_grad()；滑窗切分时按「序列 id」划分而不是按窗口划分，同一个长序列的所有窗口只能落在同一侧。最后这条是最隐蔽的，一条 30 秒的序列能切出几十个高度重叠的窗口，混在一起划分等于把训练样本复制进了验证集。

注意：出现指标异常先怀疑数据划分，再怀疑模型。十次里有七次是划分的问题，改模型往往白忙一天。

6. 二次开发：换成自己的数据，再把它做成插件

真正要落地时，改动集中在一个地方——数据适配。模型层基本不用动，只要你把特征维度对齐到FEAT_DIM和AUDIO_DIM这两个常量就行。我的做法是写一个适配器协议，把「怎么读一条序列」这件事和训练流程解耦：

from typing import Protocol class SequenceAdapter(Protocol): def list_ids(self) -> list[str]: ... def load(self, uid: str) -> tuple[np.ndarray, np.ndarray]: """返回 (motion[T,D1], audio[T,D2])，两者帧数必须一致"""

自己实现一个MyCsvAdapter，把 CSV 里的关节角度转成 6D 旋转、把音频重采样到同一帧率，剩下的滑窗、归一化、训练全都可以复用。验证顺序比写代码更重要：先拿一条序列跑单样本过拟合，loss 应该能压到 1e-3 以下，压不下去说明特征或标签有 bug，不用往下走；单样本过关了再跑全量，看验证集的重建误差；最后才是接上可视化，把 6D 输出经 Gram-Schmidt 还原成旋转矩阵、套到骨架上渲染。三步里任何一步跳过，后面都要用成倍的时间还回来。

想把它当插件用，在pyproject.toml里注册两个入口就够：一个命令行dmm = "dmmotion.cli:main"，一个给宿主程序调用的函数入口predict = "dmmotion.api:predict_sequence"，后者接收一段音频特征和初始姿态，返回逐帧动作，内部自己处理滑窗拼接和边界平滑。宿主不需要知道 Transformer 有几层，也不该知道。

最后说个习惯。早期我拿到别人给的这类「数据 + 代码」包，第一反应总是先把训练脚本跑起来，看 loss 掉了就以为成了，结果接进业务之后动作一顿一顿地抖，回头查了两天才发现是旋转表示没统一。从那以后我每次接手新数据都强制走一遍单样本过拟合加单帧可视化，宁可多花半小时，也不赌运气。希望帮到你。

本文还有配套的精品资源，点击获取