简介:面向轨迹预测任务的完整 Python 工程代码包,适用于需要开展时序建模与深度学习对比实验的学生、算法工程师或相关课题研究者。压缩包共 11 个文件,约 723KB,包含 7 个 Python 脚本、模型 checkpoint、CSV 数据集以及深度学习的索引与数据分片文件。脚本涵盖了数据读取与预处理、GRU 与 LSTM 模型实现、逻辑回归基线、轨迹数据处理与分析等功能模块;checkpoint 可恢复训练或直接评估模型,CSV 数据让复现实验更加便捷,索引文件与分片则对应常见深度学习框架的持久化格式。整体结构便于对照不同模型的训练与预测效果,也适合在真实轨迹数据集上替换或扩展算法。目前已有 230 人学习下载,适合希望快速获得可运行代码并开展轨迹预测对比实验的入门及进阶学习者。
1. 轨迹预测项目:一套完整代码包从坐标输入到未来轨迹可视化
我拆过一个叫“完整代码.zip轨迹预测”的项目包,解压之后不是只有一两个模型文件,而是一整套能跑通训练、推理、可视化的工程代码。它的核心价值很直接:把原始的历史坐标序列喂进去,就能输出未来一段时间的预测轨迹,中间的数据预处理、模型训练、评估指标、绘图脚本全都给你配好了。适合做自动驾驶、行人轨迹预测、机器人路径规划这类方向的研究者和工程师,尤其是想快速拿到一个 baseline 然后改数据、改网络结构的人。这个项目包更像一个“软件/插件”级别的工具箱,而不只是教学片段。虽然它不算大,但结构清晰,扩展起来很顺手。
2. 项目架构与核心模块:解压 zip 后你拿到的不是一坨代码
很多轨迹预测代码的痛点是“模型能跑,但数据准备和可视化一团糟”。这个项目包的优点在于它把整个流程拆成了几个职责分明的模块,我建议你拿到 zip 后先不要急着运行,而是对照目录结构把每一层的作用搞清楚。这样后面改代码时,你才知道该在哪个文件里动刀。
2.1 目录结构:data、models、utils 三层分离
打开压缩包后,你会看到类似下面的组织方式。这是我见过比较多的一种标准工程布局,适合单人开发和后期维护。
trajectory_prediction/ ├── config.py # 全局配置:路径、超参数、设备 ├── data/ │ ├── __init__.py │ ├── dataset.py # 数据加载与预处理 │ ├── raw_data/ │ │ └── sample.csv # 原始坐标序列示例 │ └── processed/ # 缓存训练样本 ├── models/ │ ├── __init__.py │ ├── encoder.py # 轨迹编码器 │ ├── decoder.py # 预测解码器 │ └── predictor.py # 把 encoder 和 decoder 串起来 ├── utils/ │ ├── metrics.py # ADE / FDE 指标 │ ├── visualize.py # 真实轨迹与预测轨迹绘图 │ └── seeds.py # 随机种子固定 ├── train.py # 训练入口 ├── evaluate.py # 评估入口 └── predict.py # 单条预测入口config.py是全局信息中心,所有可调参数都收敛到这里。你不需要在训练脚本里到处硬编码路径或学习率,改参数时只需要打开这个文件改掉对应变量。
我在拆这个项目的时候最看重的一点是:数据原始文件放在了data/raw_data/下,而处理后的训练样本会缓存到data/processed/里。这样数据预处理只做一次,后续多次调整模型参数时不会反复读原始 CSV,训练启动会很快。
2.2 数据流设计:原始坐标如何变成训练样本
原始数据格式是 CSV,每一行代表一个时间步的坐标点。它不像很多公开数据集那样直接给你切割好的序列,所以你需要自己完成“连续轨迹”到“训练样本”的转换。项目里把这个过程封装在data/dataset.py中。
# data/dataset.py 核心逻辑提取 class TrajectoryDataset(Dataset): def __init__(self, csv_path, obs_len=8, pred_len=12, stride=1): self.obs_len = obs_len self.pred_len = pred_len df = pd.read_csv(csv_path) # 这里假设 CSV 有 frame, agent_id, x, y 四列 self.segments = self._build_segments(df, stride) def _build_segments(self, df, stride): segments = [] # 按 agent_id 分组,每个 agent 的轨迹单独处理 for agent_id in df["agent_id"].unique(): agent_df = df[df["agent_id"] == agent_id].sort_values("frame") coords = agent_df[["x", "y"]].values # 滑动窗口切分输入输出对 for i in range(0, len(coords) - self.obs_len - self.pred_len + 1, stride): obs = coords[i : i + self.obs_len] pred = coords[i + self.obs_len : i + self.obs_len + self.pred_len] segments.append((obs, pred)) return segments def __len__(self): return len(self.segments) def __getitem__(self, idx): obs, pred = self.segments[idx] # 归一化时以 observation 的起点为原点 origin = obs[0] obs_norm = obs - origin pred_norm = pred - origin return ( torch.FloatTensor(obs_norm), torch.FloatTensor(pred_norm), torch.FloatTensor(pred), # 原始坐标用于评估和可视化 )这里obs_len=8表示观察过去 8 帧,pred_len=12表示预测未来 12 帧。stride是滑动步长,当轨迹点密集时可以用更大的步长来减少样本数量,避免相邻样本过于相似导致训练冗余。
这个实现有一个很关键的细节:归一化时把观测序列的第一个点作为原点。因为轨迹预测要学习的是“相对运动模式”,而不是绝对坐标。如果你把原始经纬度或者屏幕像素坐标直接丢进模型,模型很容易过拟合到特定起点,换一个场景精度立刻崩掉。以起点为原点后,模型学到的是位移趋势,泛化能力会好很多。
2.3 模型选型:LSTM 与 Transformer 融合的轨迹编码器
这个项目没有用最简单的“单层 LSTM 直出未来坐标”的方案,而是做了一层组合:观察序列先用带位置编码的 Transformer 编码器提取特征,再用一个轻量级 LSTM 作为解码器逐步生成未来点。
为什么这样选?我的理解是:Transformer 擅长捕捉长序列里点与点之间的全局依赖,而轨迹预测中当前时刻的方向会受到几步之前甚至十步之前运动模式的影响;LSTM 则擅长按时间顺序迭代生成,天然适合逐点输出未来坐标。两个结构搭配起来,既能看全局,又能顺着时间生成。
# models/predictor.py 核心定义 import torch import torch.nn as nn class TrajectoryPredictor(nn.Module): def __init__(self, input_dim=2, d_model=64, nhead=4, num_layers=2, pred_len=12): super().__init__() self.pred_len = pred_len self.input_proj = nn.Linear(input_dim, d_model) self.position_embed = nn.Parameter(torch.randn(8, 1, d_model)) encoder_layer = nn.TransformerEncoderLayer( d_model=d_model, nhead=nhead, batch_first=True ) self.encoder = nn.TransformerEncoder(encoder_layer, num_layers=num_layers) self.decoder = nn.LSTMCell(d_model, d_model) self.output_proj = nn.Linear(d_model, input_dim) def forward(self, obs): # obs shape: (batch, 8, 2) x = self.input_proj(obs) # (batch, 8, d_model) x = x + self.position_embed[:8] # 加位置编码 enc_out = self.encoder(x) # (batch, 8, d_model) last_hidden = enc_out[:, -1, :] # 取最后一个时间步的编码结果 h = last_hidden c = torch.zeros_like(h) preds = [] for _ in range(self.pred_len): h, c = self.decoder(h, (h, c)) step_out = self.output_proj(h) preds.append(step_out) return torch.stack(preds, dim=1) # (batch, pred_len, 2)input_dim=2对应归一化后的 (x, y) 坐标。d_model=64是特征维度,调大可以增强模型容量,但训练显存也会上去。nhead=4表示多头注意力头数,一般取 2 的倍数。num_layers是 Transformer 编码器层数。
在解码部分,每次 LSTM Cell 输入是上一次的隐藏状态,输出被投影成 2 维坐标。这种做法常见于自回归生成模型,每一步都依赖上一步的内部状态,因此预测轨迹是平滑连续的。如果直接让 Transformer 一次性输出 12 个点,轨迹容易产生跳变,这是我在别的项目里踩过的坑。
3. 核心代码拆解:从数据加载到训练循环的完整链路
上一章讲的都是模块划分和设计思路,这一章我会贴出项目里真正驱动训练和推理的那几段代码。你后续改模型也好、换数据也好,主要打交道的就是这些文件。
3.1 数据加载与特征工程:滑动窗口与坐标归一化
在dataset.py里,除了前面看到的切分逻辑,还做了一个比较细的处理:对每个 agent 的轨迹按 frame 排序,遇到缺失帧或长时间静止片段会跳过或切断。这一点很多纯教学代码不会考虑,但真实数据里几乎一定会出现传感器丢帧、目标静止不动的情况。
# data/dataset.py 中补充的异常处理逻辑 def _build_segments(self, df, stride): segments = [] for agent_id in df["agent_id"].unique(): agent_df = df[df["agent_id"] == agent_id].sort_values("frame") coords = agent_df[["x", "y"]].values # 如果轨迹总长不够,直接跳过 if len(coords) < self.obs_len + self.pred_len + 1: continue # 用帧差检测时间不连续的位置,切分子轨迹 frame_diff = np.diff(agent_df["frame"].values) cut_points = np.where(frame_diff > 2)[0] + 1 for cut in cut_points: sub1 = coords[:cut] sub2 = coords[cut:] # 对每一段分别做滑窗,合并结果 segments.extend(self._window_a_sub_traj(sub1)) segments.extend(self._window_a_sub_traj(sub2)) return segments def _window_a_sub_traj(self, sub_coords): segs = [] if len(sub_coords) < self.obs_len + self.pred_len: return segs for i in range(0, len(sub_coords) - self.obs_len - self.pred_len + 1, 1): obs = sub_coords[i : i + self.obs_len] pred = sub_coords[i + self.obs_len : i + self.obs_len + self.pred_len] segs.append((obs, pred)) return segs这里把帧差大于 2 的位置视为轨迹断裂,断裂前后的轨迹不能拼在一起做预测,否则模型会把两个不相关运动阶段当成连续轨迹来学,训练出来的结果会非常不稳定。我自己在跑这个项目时遇到过类似问题——原本精度看起来还行,后来换了数据源后 loss 一直在高位震荡,检查后才发现是缺失帧造成的假样本。
所以只要是轨迹预测类任务,数据加载阶段一定要考虑时间连续性。滑动窗口的切分参数obs_len和pred_len也要根据实际帧率来定,比如你拿到的视频是 30 帧每秒,那观察 8 帧就是约 0.27 秒的运动历史,预测 12 帧就是未来 0.4 秒。这决定了模型能“看”多远的过去和预测多远的未来,不能盲目照搬。
3.2 模型训练:损失函数、优化器和训练入口
项目里的train.py是整个流程的驱动脚本。它读取配置、初始化模型和数据、启动训练循环,并且每隔若干轮会在验证集上算一次平均位移误差(ADE)和最终位移误差(FDE)。
# train.py 核心训练逻辑简化 def train_one_epoch(model, dataloader, optimizer, device): model.train() total_loss = 0.0 for obs, pred, _ in dataloader: obs, pred = obs.to(device), pred.to(device) optimizer.zero_grad() pred_out = model(obs) # 坐标预测用均方差损失 loss = F.mse_loss(pred_out, pred) loss.backward() optimizer.step() total_loss += loss.item() * obs.size(0) return total_loss / len(dataloader.dataset) # 主循环 model = TrajectoryPredictor( input_dim=2, d_model=64, nhead=4, num_layers=2, pred_len=12 ) optimizer = torch.optim.Adam(model.parameters(), lr=1e-3, weight_decay=1e-5) scheduler = torch.optim.lr_scheduler.StepLR(optimizer, step_size=30, gamma=0.5) for epoch in range(100): train_loss = train_one_epoch(model, train_loader, optimizer, device) scheduler.step() if epoch % 10 == 0: ade, fde = evaluate_on_val(model, val_loader, device) print(f"Epoch {epoch}: train_loss={train_loss:.4f}, ADE={ade:.2f}, FDE={fde:.2f}")ADE是预测轨迹和真实轨迹在所有时间步上的平均欧氏距离,FDE只看最后一个预测点与真实终点的距离。业界评估轨迹预测模型通常这两个指标一起看:ADE 衡量整体贴合度,FDE 衡量最终停点误差。如果你的应用场景是“提示司机前车会不会变道”,FDE 比 ADE 更重要;如果是要跟踪整个运动趋势,ADE 更关键。
优化器采用了 Adam,使用StepLR每 30 轮衰减一半学习率。这种衰减方式在轨迹序列预测任务里比较省心,不用手动调每个 epoch 的学习率,避免后期在小幅度 loss 上来回震荡。weight_decay 设得比较小,因为轨迹预测模型参数本来就少,正则太强反而欠拟合。
3.3 评估与推理:预测一条完整轨迹
项目里的predict.py单独承担“给一段历史坐标,输出未来轨迹”的职责。它跟训练脚本最大的区别在于:预测阶段要做反归一化,把网络输出的以起点为原点的相对坐标转换成真实世界坐标。
# predict.py 单条推理核心逻辑 def predict_trajectory(model, history, device): model.eval() # history: numpy array with shape (obs_len, 2),原始坐标 origin = history[0] norm_history = torch.FloatTensor(history - origin).unsqueeze(0).to(device) with torch.no_grad(): norm_future = model(norm_history) # 反归一化:加上起点坐标 future = norm_future.cpu().numpy().squeeze(0) + origin return future注意这里model.eval()是必须的,虽然这个模型结构里没有 dropout 和 batchnorm,但养成习惯总是好事。我把这个反归一化单独拎出来强调,是因为很多人在训练时不构造原始坐标返回的 sample,导致预测的时候只能得到归一化后的轨迹,画图时对着一个相对坐标系干瞪眼。
predict_trajectory返回的future是(pred_len, 2)的数组,表示未来 12 个时间步的绝对坐标。在这个项目中,你只需要把模型输出的pred_norm加上origin就是真实轨迹。这里没有做坐标缩放,如果数据里包含不同量纲的特征,还需要在dataset.py里记录每组样本的均值和标准差,预测时再做完整反归一化。
4. 跑通实验:从环境配置到可视化结果
这一章面向第一次拿到 zip 就想把模型跑起来的同学。我尽量把环境、命令和输出讲清楚,你不需要理解每一行代码也能先看到预测效果,然后再回头研究实现。
4.1 环境依赖:Python 版本与第三方库
根据项目包里的requirements.txt,核心依赖是这几个:
pytorch>=1.9 numpy>=1.20 pandas>=1.2 matplotlib>=3.3建议使用 Python 3.8 以上版本,因为torch.FloatTensor等 API 在旧版本上兼容性不好。如果是在 GPU 环境,还需要提前装好对应版本的 CUDA 版 PyTorch。
# 创建虚拟环境并安装依赖 python -m venv traj_env source traj_env/bin/activate # Windows 下是 traj_env\Scripts\activate pip install -r requirements.txt4.2 训练脚本:命令行参数与运行方式
项目没有用 argparse 去接收全部参数,而是把参数写在config.py里,训练时直接执行python train.py。这样避免了命令行参数太多导致手滑打错字,也方便用配置文件管理多组实验。
python train.py如果你想快速验证代码能否跑通,可以把config.py里的num_epochs调小到 5,batch_size调小到 16。我一般会先跑一个小 epoch 数确认数据加载和 loss 计算没问题,再开启正式训练。训练过程中控制台会每 10 个 epoch 输出一次 ADE 和 FDE,如果这两个指标在下降,说明模型真的在学到东西。
4.3 可视化结果:真实轨迹和预测轨迹对比
utils/visualize.py提供了一个简单的绘图函数,它会读取data/raw_data/sample.csv中某个 agent 的历史轨迹和真实未来轨迹,再把模型预测结果画在同一张图上。
# utils/visualize.py 画图核心 import matplotlib.pyplot as plt def visualize_prediction(history, pred_future, gt_future, save_path=None): plt.figure(figsize=(6, 6)) plt.plot(history[:, 0], history[:, 1], "b-", label="History") plt.plot(gt_future[:, 0], gt_future[:, 1], "g--", label="Ground Truth") plt.plot(pred_future[:, 0], pred_future[:, 1], "r--", label="Prediction") plt.legend() plt.xlabel("x") plt.ylabel("y") if save_path: plt.savefig(save_path, dpi=120) else: plt.show()画图时需要注意,history和pred_future传递的都是原始绝对坐标。如果你在dataset.py里做了归一化但没有反归一化,画出来的轨迹起点会漂移到原点附近,看起来根本对不上,这一点后面避坑章节会细讲。
运行python predict.py --agent_id 1就能看到某条具体样本的预测曲线。蓝色是已知历史,绿色虚线是未来真实轨迹,红色虚线是模型预测。如果红色和绿色贴合程度较高,说明模型在这个样本上表现不错。
5. 常见问题与避坑指南:我在改这个项目时踩过的坑
这个项目整体比较完善,但只要你把自己的数据集换进去或者修改网络结构,各种各样的问题就会冒出来。我整理了几个最容易踩中的坑,每一条都是实际体验过的,不是只存在于文档里。
5.1 归一化了却忘记反归一化
现象:训练时 loss 降到很低,但预测出来的轨迹画在图上一堆点挤在原点附近,或者尺度完全不对。
原因:模型在归一化坐标系下训练,预测输出也是以观测起点为原点的相对坐标。如果你在推理时直接把这个输出当作绝对坐标使用,就会看到轨迹偏离真实位置。
解决:在评估和推理代码中,把预测结果加上每个样本的origin。这个origin必须是训练时减掉的那个起点,不能拿整条轨迹的均值或者终点来加。项目里dataset.py已经返回了原始坐标pred,所以推理时直接用pred_norm + origin即可。
5.2 序列长度设置不当导致“看不够”或“过度拟合”
现象:预测精度忽高忽低,改了obs_len或pred_len后结果非常不稳定。
原因:obs_len决定模型能看到多长的历史信息。如果你的轨迹数据是低频采样,比如每秒只有 2 个点,那么obs_len=8只代表 4 秒历史,可能不够模型判断运动趋势。pred_len太长则会让模型学习一个难以预测的长距离输出,loss 会集中在远距离点上。
解决:根据实际帧率调整obs_len和pred_len。我一般会让观察时长覆盖至少 2 到 3 秒的轨迹,预测时长不要超过观察时长的 1.5 倍。你可以在config.py里同时保存这两个参数,方便不同数据集之间切换。调整后重新生成数据缓存,否则旧的processed数据还会沿用之前的切分方式,造成训练和预测不一致。
5.3 显存不足的翻车现场
现象:训练到中途被CUDA out of memory打断,batch size 再往下调还是报错。
原因:Transformer 编码器的显存占用随num_layers和nhead递增,再加上序列长度,显存消耗比普通 LSTM 高不少。另外,如果训练时开了torch.autograd.set_detect_anomaly(True),显存占用会进一步暴涨。
解决:先把batch_size降到 8 或 4,如果还不行就减小d_model,从 64 降到 32 通常能省出不少显存。梯度累积也是一个后悔药:在训练循环里累积多个 batch 的梯度再更新参数,效果上等价于更大 batch,显存却能降下来。
optimizer.zero_grad() for i, (obs, pred, _) in enumerate(dataloader): loss = compute_loss(obs, pred) / accumulate_step loss.backward() if (i + 1) % accumulate_step == 0: optimizer.step() optimizer.zero_grad()注意这里 loss 要除以累积步数,否则实际梯度会跟原始 batch 大小不匹配,学习率相当于被不小心放大了。
5.4 不固定随机种子,复现结果是玄学
现象:同一次训练跑两遍,最终的 ADE 和 FDE 完全不一样,有时候差别还不小。
原因:模型初始化、DataLoader 的随机打乱顺序、甚至 PyTorch 后端算子都有可能引入随机性。如果你不固定随机种子,实验结果就丧失了可复现性,后续改代码时也分不清指标提升是来自改的结构还是碰巧的随机初始化。
解决:在train.py最前面固定所有能固定的随机源。
import torch import numpy as np import random def set_seed(seed): random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) torch.backends.cudnn.deterministic = True torch.backends.cudnn.benchmark = False set_seed(42)cudnn.deterministic = True会让卷积和注意力里的算法选择固定路径,虽然会损失一些性能,但换来的是可复现的结果。注意 CUDNN 的 benchmark 要关掉,否则它会动态选择最快的卷积算法,算法不同也会引入随机差异。
5.5 可视化时坐标系不一致
现象:画出来的历史轨迹和预测轨迹出现上下颠倒或者偏转,看起来像是模型“反着走”。
原因:常见于经纬度坐标直接用绘图库的横纵轴,不同数据源的 y 轴方向可能相反,比如图像坐标系 y 向下,地图坐标系 y 向上。
解决:在可视化脚本里统一坐标方向,或者直接将数据坐标做一次线性变换,让所有轨迹统一到“x 向右、y 向上”的坐标系。遇到这种情况不要慌,分析模型输出先看数值变化趋势,再看画图效果。这个坑虽然傻瓜,但真的会浪费一晚上。
6. 二次开发方向:把通用轨迹模型改成你自己的专属预测器
当你跑通了原项目、理解了每一行代码之后,就可以开始动工了。这一章我不讲泛泛的“可以加注意力”之类的话,而是给出三个具体的改造方向,每个方向都配有操作思路和注意事项。
第一个方向是换数据源。原始项目给了sample.csv,但你要用的可能是 NuScenes、SDD 或者其他平台导出的数据。改造时不要直接重写dataset.py,而是新增一个适配器函数,把外部数据转换成项目要求的frame, agent_id, x, y四列。比如从 JSON 中读取轨迹,可以写一个json_to_csv()方法,保留原来的TrajectoryDataset不变,只改数据入口。这样当你要回退到原始数据时,切回sample.csv就够了。
第二个方向是多模态轨迹预测。目前的实现是单确定性的,每次预测只输出一条轨迹。真实场景中人的意图有多重可能性,你可以在解码部分加一个 CVAE 或扩散过程,让模型输出多条候选轨迹。最简单的思路是让 LSTM 解码器接受一个噪声向量作为额外输入,生成多个未来轨迹,训练时用最小ADE来选择最优的那条。这个改造涉及predictor.py和损失函数,但如果你已经熟练了上一章的代码,改动并不复杂。
第三个方向是部署到生产环境。训练好的 PyTorch 模型可以导出为 ONNX 格式,这样在 C++ 或移动端上也能跑。导出时需要注意把动态轴和批量维度定义清楚,当时的代码里我用了固定 batch size,导出就没那么顺。
# 导出 ONNX 示例 import torch.onnx def export_onnx(model, output_path): model.eval() dummy_input = torch.randn(1, 8, 2) torch.onnx.export( model, dummy_input, output_path, input_names=["obs"], output_names=["pred"], dynamic_axes={"obs": {0: "batch"}, "pred": {0: "batch"}}, opset_version=11, )导出后可以在 Python 里用onnxruntime加载测试一下性能,确认跟 PyTorch 结果一致。这一步不算难,但它能让你把模型从实验环境搬到真实业务系统里,比如把这个预测模块嵌入到自动驾驶的决策管线中作为前置感知组件。
我在做这类二次开发时养成了一个习惯:每改一个功能,就先把原项目跑通得到一组 baseline 指标,再动代码,改完用同样的验证集对比。不对比就上线等于拿生产环境当试验田,教训够多了。从那以后我每次更换数据、调整网络结构前都强制走一遍“跑通原项目 → 记录指标 → 改代码 → 回归对比 → 更新指标”的循环,节奏慢一点,但能保证每一步改进都是真实的。这条习惯救过我好几次,希望也能帮到你。
本文还有配套的精品资源,点击获取