简介面向具备Python与深度学习基础、希望实践轨迹预测并具备一定代码阅读能力的学生和算法开发者这套压缩包提供可直接研读的完整工程代码及配套数据。资源共11个文件以7个Python脚本为核心辅以模型检查点、一份CSV轨迹数据以及TensorFlow数据分片与索引文件整体约723KB体量轻便但模块相对完整。代码围绕轨迹数据处理、LSTM与GRU循环神经网络建模展开并加入逻辑回归作为对比基线覆盖数据读取、模型定义、训练状态保存与结果分析等环节能够帮助读者理解序列预测项目的组织方式与实验流程。已有231人学习适合用作课程设计、毕业项目或小规模实验参考也可替换数据、调整网络结构并复现对比实验进一步掌握从预处理、训练到评估的完整思路。1. 拆开完整代码.zip轨迹预测先跑通哪条最小链路完整代码.zip轨迹预测这个压缩包最容易被误当成“跑通就完事”的课程作业实际拆开后真正决定效果的是数据窗口、坐标系和评估口径而不是模型层数。它解决的是从历史轨迹推未来若干帧的完整闭环原始坐标清洗、滑动窗口采样、序列模型训练、ADE/FDE 评估、可视化再到推理软件/插件入口。适合手里有轨迹表、想快速搭一个可复现基线的人也适合已经跑过模型、但卡在归一化泄漏和上线接口上的熟手。先别急着改网络按数据形状、scaler、模型输出、反归一化、指标这五步跑通最小链路后面换模型才有对照。2. 轨迹数据预处理把原始坐标压成滑动窗口的四个参数2.1 先约定字段、坐标系与张量形状轨迹预测代码包翻车十有八九不是模型结构而是字段顺序和坐标系没对齐。原始表里常见字段是object_id、frame_id、timestamp、x、y、scene_id、valid但有的包把x/y写成lon/lat有的把像素坐标和米制坐标混在一起。进模型前我一般统一成局部米制坐标以场景起点或场景中心做平移保留同一场景内的相对位置不直接把经纬度喂给序列模型。输入张量固定为[B, T_obs, 2]输出为[B, T_pred, 2]另带[B, T_obs]的有效性 mask如果后续要包成软件/插件推理入口的字段名必须和训练脚本完全一致不能靠“到时候再映射”。先跑一个字段检查脚本确认排序、重复帧和缺失比例再动转换脚本。下面这段只读原始表不生成样本目的是把黑匣子打开。import pandas as pd # 读取原始轨迹表列名按实际文件调整 df pd.read_csv(data/raw.csv) # 必要字段检查缺一个就先补字段映射 required [object_id, frame_id, timestamp, x, y, scene_id] missing [c for c in required if c not in df.columns] assert not missing, f缺少字段: {missing} # 同一对象按时间排序检查重复帧和坐标缺失 df df.sort_values([scene_id, object_id, timestamp]).reset_index(dropTrue) dup df.duplicated([scene_id, object_id, frame_id]).sum() na_ratio df[[x, y]].isna().mean().to_dict() print(重复帧数量:, dup) print(坐标缺失比例:, na_ratio) print(df[[scene_id, object_id, frame_id, x, y]].head())这段检查的逻辑很直接先保证字段齐全再按场景、对象、时间排序重复帧和缺失比例决定后面是丢弃、插值还是重采样。required里的字段名按压缩包内实际 CSV 替换但语义不要改scene_id必须参与排序和切分否则同一场景的轨迹会被随机拆到训练集和验证集指标会虚高。2.2 用转换脚本生成滑动窗口样本并固化 scaler轨迹预测的样本不是“一行一个未来点”而是“一段历史 一段未来”。我一般用obs_len8、pred_len12、stride1起步按对象切滑动窗口窗口内有效点比例低于 0.8 就丢弃。归一化只使用训练集的均值和方差验证集、测试集和推理插件都复用同一份 scaler这一步如果偷懒对全量数据fit后面评估会好看得离谱上线就露出原形。下面脚本把原始表转成npz同时保存scaler.json和meta.json。代码里按场景切分避免同一轨迹泄漏。import json import numpy as np import pandas as pd OBS_LEN, PRED_LEN, STRIDE 8, 12, 1 MIN_VALID_RATIO 0.8 def build_windows(group): group group.sort_values(timestamp) xy group[[x, y]].to_numpy(dtypenp.float32) valid group[valid].to_numpy(dtypenp.float32) if valid in group else np.ones(len(group), dtypenp.float32) samples [] for start in range(0, len(group) - OBS_LEN - PRED_LEN 1, STRIDE): obs xy[start:start OBS_LEN] pred xy[start OBS_LEN:start OBS_LEN PRED_LEN] obs_mask valid[start:start OBS_LEN] if obs_mask.mean() MIN_VALID_RATIO: continue # 以最后一个观测点为原点做相对位移减少绝对坐标漂移 origin obs[-1:].copy() samples.append((obs - origin, pred - origin, obs_mask)) return samples df pd.read_csv(data/raw.csv).sort_values([scene_id, timestamp]) train_scenes sorted(df[scene_id].unique())[: int(df[scene_id].nunique() * 0.7)] train_df df[df[scene_id].isin(train_scenes)] all_samples [] for _, g in train_df.groupby([scene_id, object_id]): all_samples.extend(build_windows(g)) obs_arr np.stack([s[0] for s in all_samples]) pred_arr np.stack([s[1] for s in all_samples]) mask_arr np.stack([s[2] for s in all_samples]) # 只用训练集统计量保存给验证、测试和插件复用 mean obs_arr.reshape(-1, 2).mean(axis0) std obs_arr.reshape(-1, 2).std(axis0) 1e-6 np.savez(data/processed/train.npz, obs(obs_arr - mean) / std, pred(pred_arr - mean) / std, maskmask_arr) with open(data/processed/scaler.json, w) as f: json.dump({mean: mean.tolist(), std: std.tolist()}, f) with open(data/processed/meta.json, w) as f: json.dump({obs_len: OBS_LEN, pred_len: PRED_LEN, stride: STRIDE}, f)这里的参数要按采集频率和任务时限调整obs_len决定模型能看到多长历史pred_len决定要推多远stride控制样本重叠度。origin取最后一个观测点做相对平移是为了让模型学“接下来往哪走”而不是记绝对坐标mask保留无效点训练时不要直接当零坐标用。mean/std必须写入文件推理插件加载模型时一起加载否则脚本和插件会出现同一段输入两个结果。2.3 四个参数怎么定窗口、步长、归一化和缺失值窗口长度不是越长越好。历史太短模型看不到转向意图历史太长早期帧和未来帧相关性弱还会放大噪声。我一般从 8 帧历史、12 帧未来起步采样频率固定后先看匀速基线能不能跑出合理误差再决定是否加长。步长stride在训练集可以取 1 增加样本验证集不要重叠采样否则指标会被相似窗口拉平。参数起始值调整依据常见误用obs_len8采集频率、转向持续时间把不同帧间隔混在一个窗口pred_len12业务预警提前量训练和评估长度不一致stride1样本量与过拟合验证集也重叠采样min_valid_ratio0.8缺失率与插值质量缺失点直接填 0缺失值处理要分场景短缺口可以线性插值长缺口宁可丢窗口不要用均值硬填。归一化统计量只来自训练集验证和测试按同一份mean/std变换如果包里的脚本对全量数据fit先改掉再谈模型。自检时随机抽三个窗口做一次反归一化确认预测点和真实点还在同一坐标系里这一步比看 loss 更能提前发现翻车。提示转换完成后先打印obs_arr.shape、pred_arr.shape和mask_arr.mean()形状对不上时不要继续训练后面所有指标都会错位。3. 模型训练与损失函数GRU 基线怎么接进完整代码3.1 为什么先用编码器-解码器基线轨迹预测的模型选型很容易一上来就追新结果连数据管线都没跑顺。我的习惯是先接一个 GRU 编码器-解码器编码器吃历史轨迹解码器逐帧输出未来位移输入输出形状和后面的 Transformer、时序卷积都能对齐。它训练快、显存低、可解释性强适合当完整代码包的第一条基线。模型预测相对最后一个观测点的位移而不是绝对坐标这样换场景时漂移小解码器输出[B, T_pred, 2]损失只在有效未来点上计算。import torch import torch.nn as nn class GRUTrajectoryBaseline(nn.Module): def __init__(self, hidden_dim64, pred_len12): super().__init__() self.pred_len pred_len self.encoder nn.GRU(input_size2, hidden_sizehidden_dim, batch_firstTrue) self.decoder nn.GRU(input_size2, hidden_sizehidden_dim, batch_firstTrue) self.head nn.Linear(hidden_dim, 2) def forward(self, obs, pred_gtNone): # obs: [B, T_obs, 2]已经做过相对平移和归一化 _, hidden self.encoder(obs) dec_input obs[:, -1:, :] # 用最后一帧启动解码 outputs [] for _ in range(self.pred_len): out, hidden self.decoder(dec_input, hidden) step self.head(out) # [B, 1, 2] outputs.append(step) dec_input step return torch.cat(outputs, dim1)这段模型只做最小闭环编码器提取历史状态解码器逐步生成未来位移head把隐藏状态映射回二维坐标。hidden_dim从 64 或 128 起步pred_len必须和预处理保存的meta.json一致推理时不要传pred_gt训练时也不要用真实未来帧喂给解码器否则验证阶段会高估效果。3.2 训练循环、端点加权和模型保存轨迹预测的平均损失容易掩盖终点漂移我一般在主损失外加一个端点加权项让最后一帧的误差多承担一些权重。训练循环里保留梯度裁剪、验证集反归一化评估和最佳模型保存保存的不只是state_dict还要带config和scaler否则推理插件加载后还得靠人脑记参数。import json import numpy as np import torch from torch.utils.data import DataLoader, TensorDataset def train_one_epoch(model, loader, optimizer, device, endpoint_weight0.5): model.train() total 0.0 for obs, pred, mask in loader: obs, pred obs.to(device), pred.to(device) optimizer.zero_grad() out model(obs) loss_all torch.nn.functional.smooth_l1_loss(out, pred, reductionnone) loss loss_all.mean() # 最后一帧误差加权缓解终点漂移 loss loss endpoint_weight * loss_all[:, -1].mean() loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm5.0) optimizer.step() total loss.item() * obs.size(0) return total / len(loader.dataset) # 数据加载和训练参数集中在一处方便插件复用 cfg {hidden_dim: 64, batch_size: 128, lr: 1e-3, epochs: 50, endpoint_weight: 0.5} data np.load(data/processed/train.npz) loader DataLoader(TensorDataset(torch.tensor(data[obs]), torch.tensor(data[pred]), torch.tensor(data[mask])), batch_sizecfg[batch_size], shuffleTrue) model GRUTrajectoryBaseline(hidden_dimcfg[hidden_dim], pred_lendata[pred].shape[1]) optimizer torch.optim.Adam(model.parameters(), lrcfg[lr]) for epoch in range(cfg[epochs]): loss train_one_epoch(model, loader, optimizer, cpu, cfg[endpoint_weight]) print(epoch, round(loss, 6)) torch.save({state_dict: model.state_dict(), cfg: cfg}, checkpoints/gru_baseline.pt)endpoint_weight控制终点误差的权重太大容易让中间帧欠拟合太小则终点漂移明显0.3 到 1.0 之间先扫一轮。clip_grad_norm_是序列模型的后悔药梯度爆炸时先看它有没有生效。保存字典里带cfg是为了让推理软件/插件加载时不用猜hidden_dim和pred_len如果包里只保存了裸state_dict建议补一个config.json后面排错会省很多时间。3.3 超参和显存不够时怎么改显存不够时优先降batch_size和hidden_dim不要先砍pred_len因为未来长度是业务契约训练和评估必须一致。学习率从1e-3起步验证损失震荡就降到5e-4或加 warmup过拟合表现为训练损失继续降、验证 ADE 不降先加 dropout 或减少历史窗口再考虑换模型。下面这张表是我在轨迹预测基线里常用的调参顺序。现象优先调整次选不要做训练 loss 不降学习率、数据归一化隐藏维度盲目加层验证 ADE 反弹dropout、早停减小窗口用测试集调参终点误差大endpoint_weight加终点损失只改可视化显存不足batch_sizehidden_dim改pred_len训练脚本跑通后先用匀速基线对比一次拿最后一个观测点的速度外推未来算同一个 ADE/FDE。如果 GRU 基线连匀速都打不过问题通常不在模型容量而在坐标、窗口或评估口径。把这条对照写进日志后面换更复杂的模型才有意义。4. 评估与可视化ADE/FDE 之外的三个验收指标4.1 指标口径先统一再谈模型好坏ADE 是未来各帧位移误差的平均FDE 是最后一帧误差两者都必须在反归一化之后算单位要和业务坐标一致。很多代码包在归一化空间里算指标数值看着很小但没有物理意义还有的把无效帧也平均进去缺失越多指标越好看。我一般再补三个验收项Miss Rate终点误差超过阈值的样本比例、终点误差 P90、速度/加速度越界率。前者看极端失败后两者看轨迹是否物理合理。指标计算口径用途ADE未来帧 L2 误差均值整体拟合FDE最后一帧 L2 误差终点精度MRFDE 大于阈值的比例极端失败P90终点误差 90 分位长尾风险越界率速度/加速度超阈值比例物理合理性阈值不要拍脑袋按场景的采样间隔和对象类型定同一个包里训练和评估的阈值必须来自同一份配置。评估集按场景切分不能把同一对象的相邻窗口拆到两边否则指标会虚高。若包内脚本只输出一个平均 ADE先补上按场景、按对象、按预测步长的拆分否则你无法判断模型是整体差还是某几帧崩。4.2 评估脚本反归一化后再累计误差评估脚本要先把预测和真实值反归一化再按样本累计误差不能先算 batch 平均再平均 batch。下面的写法保留每个样本的 ADE/FDE方便后面算 P90 和 Miss Rate。import json import numpy as np import torch def evaluate(model, loader, scaler, devicecpu, miss_threshold2.0): model.eval() mean torch.tensor(scaler[mean], dtypetorch.float32, devicedevice) std torch.tensor(scaler[std], dtypetorch.float32, devicedevice) ade_list, fde_list [], [] with torch.no_grad(): for obs, pred, mask in loader: obs, pred obs.to(device), pred.to(device) out model(obs) # 反归一化回米制坐标再算物理误差 out out * std mean pred pred * std mean dist torch.norm(out - pred, dim-1) # [B, T_pred] ade_list.extend(dist.mean(dim1).cpu().numpy()) fde_list.extend(dist[:, -1].cpu().numpy()) ade np.array(ade_list) fde np.array(fde_list) return { ADE: float(ade.mean()), FDE: float(fde.mean()), MR: float((fde miss_threshold).mean()), FDE_P90: float(np.percentile(fde, 90)), } with open(data/processed/scaler.json) as f: scaler json.load(f) metrics evaluate(model, loader, scaler) print(metrics)miss_threshold要按业务容差设单位是米或像素取决于坐标系FDE_P90比平均值更能暴露长尾失败。model.eval()和torch.no_grad()不是装饰推理插件里漏掉它们dropout 和 BN 会让同一输入两次结果不同。指标输出建议写进 JSON 日志和模型文件、scaler、数据版本放在一起后面复盘才有依据。4.3 可视化和基线对比别只看 loss 曲线可视化不是画着好看而是检查轨迹方向、尺度和平移有没有错。我一般抽三类样本正常直行、转弯、终点误差最大的失败样本。每张图画历史轨迹、真实未来、模型预测和匀速基线坐标轴保持等比例否则转弯会被视觉拉直。若预测整体偏移一个固定向量优先查相对原点和平移若方向反了查x/y顺序和坐标系旋转。import matplotlib.pyplot as plt def plot_case(obs, pred, out, path): # obs/pred/out 均为反归一化后的 [T, 2] plt.figure(figsize(5, 5)) plt.plot(obs[:, 0], obs[:, 1], o-, labelhistory) plt.plot(pred[:, 0], pred[:, 1], s-, labelground truth) plt.plot(out[:, 0], out[:, 1], x--, labelprediction) plt.axis(equal) plt.legend() plt.savefig(path, dpi150, bbox_inchestight) plt.close()画图前先反归一化axis(equal)必须加不然横向纵向比例不同会误导判断。把匀速基线画在同一张图上能快速看出模型是否只是在学“沿最后方向直走”。如果模型和基线几乎重合先别加注意力回去检查历史窗口里有没有速度变化信息如果模型在转弯样本上明显滞后再考虑加转向特征或换时序结构。5. 避坑与排查轨迹预测代码包落地时的五类翻车5.1 五条踩坑记录现象、原因、解决第一条训练 loss 很低但评估 ADE 爆炸。现象是训练日志漂亮验证集误差大一个数量级原因通常是对全量数据做了归一化或者评估时忘了反归一化。解决是把scaler固化到训练集评估脚本先乘std加mean再算米制误差脚本和插件共用同一份scaler.json。第二条验证集好得不真实。现象是验证 ADE 接近训练集换个场景就崩原因是按样本随机切分同一对象的相邻窗口同时进了训练和验证。解决是按scene_id或object_id切分验证集不重叠采样测试集单独留场景。第三条推理插件和脚本结果对不上。现象是同一段轨迹在脚本里正常包成软件/插件后偏移或抖动原因多是字段顺序、eval()、no_grad()、相对原点不一致。解决是把预处理、模型前向、反归一化封成一个类脚本和插件都调这个类输入输出用同一份契约测试。第四条batch 推理出现 NaN 或形状错误。现象是单条推理正常多条一起就报错原因是轨迹长度不一、空轨迹、mask 没对齐。解决是推理前按有效长度排序或 padding空输入直接返回明确错误码不要用零坐标硬凑。第五条可视化和指标错位。现象是图上看着还行指标却很差原因是frame_id没排序、重复帧没去重、时间戳和帧号混用。解决是转换阶段固定排序键重复帧按规则合并评估和画图都从同一份npz读数据。注意这五类问题里前三类会直接让结论失真后两类会让排错方向跑偏。先修数据契约再调模型。5.2 排查顺序与日志字段排查不要从模型结构开始我一般按“形状 → 坐标系 → scaler → 模型状态 → 反归一化 → 指标”顺序走。每一步打一条结构化日志字段包括scene_id、object_id、obs_shape、pred_shape、scaler_mean、scaler_std、model_mode、ade_raw、ade_denorm。这样脚本和插件对不上时能一眼看出是哪一步分叉。import json import logging logging.basicConfig(levellogging.INFO, format%(message)s) def log_once(tag, obs, pred, scaler, model): record { tag: tag, obs_shape: list(obs.shape), pred_shape: list(pred.shape), scaler_mean: scaler[mean], scaler_std: scaler[std], model_training: model.training, has_nan: bool(obs.isnan().any() or pred.isnan().any()), } logging.info(json.dumps(record, ensure_asciiFalse))model_training能直接暴露插件里漏掉eval()的问题has_nan放在预处理之后、模型之前能区分是输入坏还是前向坏。日志不要只打 loss形状、scaler、模式这三个字段最能定位分叉。每次改预处理或换模型先用固定小样本跑一遍脚本和插件的一致性对比误差超过1e-5就停下来查不要带病上线。6. 插件化推理与验收把预测器包成可调用组件的具体技巧软件/插件形态落地时我最看重三件事输入契约、预处理与模型绑定、失败可观测。输入契约固定为obs: [T,2]、mask: [T]、meta输出为pred: [T_pred,2]和status预处理、模型前向、反归一化封进同一个类脚本和插件都调它避免两套代码各走各的。下面这个最小类可以直接嵌到服务入口里加载时把scaler和config一起读进来。import json import torch class TrajectoryPredictor: def __init__(self, ckpt_path, scaler_path, devicecpu): ckpt torch.load(ckpt_path, map_locationdevice) self.model GRUTrajectoryBaseline(**ckpt[cfg]) self.model.load_state_dict(ckpt[state_dict]) self.model.eval() self.device device with open(scaler_path) as f: s json.load(f) self.mean torch.tensor(s[mean], dtypetorch.float32, devicedevice) self.std torch.tensor(s[std], dtypetorch.float32, devicedevice) torch.no_grad() def predict(self, obs): x torch.tensor(obs, dtypetorch.float32, deviceself.device).unsqueeze(0) if torch.isnan(x).any(): return None, INPUT_NAN out self.model(x)[0] out out * self.std self.mean # 回到米制坐标 return out.cpu().numpy(), OKckpt[cfg]必须包含模型构造参数scaler_path必须和训练时同一份文件torch.no_grad()和eval()是插件稳定性的底线。验收时至少跑空输入、NaN 输入、单条、批量、脚本与插件同输入对比五项终点最大差值超过1e-5就回查字段顺序和反归一化。从那以后我每次把轨迹预测包成软件/插件都强制走一遍脚本与插件同输入对比差一点就停工。希望帮到你。本文还有配套的精品资源点击获取
