把 Foundation Model 用到空域预测真正要解决的不是“训练一个超大模型”而是先把空域问题转换成模型能吃到的数据形态。空域预测在民航流量管理、无人机物流、城市空中交通里很常见目标通常是根据过去几十分钟到几小时的飞行轨迹、航班计划和气象条件推断未来 15 分钟、30 分钟甚至更长时间内某个区域的飞行密度、拥堵概率和容量余量。对于刚接触这个方向的人来说最容易犯的错误是一上来就搜“空域预测模型”却忽略了数据组织、任务定义和评估方式。这三件事决定了模型能不能落地也决定了最终代码是变成一个可维护的工程还是停留在实验脚本。这篇文章会按“任务定义 - 数据准备 - 模型设计 - 训练验证 - 排错优化”的顺序用一个轻量可运行的 PyTorch 示例讲清楚如何搭建一个预测空域状态的基础模型原型。读者不需要有航空背景但最好了解 Python、PyTorch 和深度学习基础。文中的代码可以用 CPU 跑通适合用来理解整个链路也为后续接入真实空域数据打一个骨架。1. 先重新定义任务基础模型到底预测空域的什么1.1 空域预测的三个典型输出空域预测不是一个单一任务至少需要区分三类输出输出类型含义典型用途常见粒度密度某个空域网格内目标数量或并发架次判断拥挤程度、热点识别网格、扇区容量当前空域还能接纳多少飞行活动辅助流量管理决策扇区、航路点流量单位时间通过某断面或航路点的架次预战术流量预测航路点、断面这三类输出可以共用大部分输入数据但任务定义和损失函数不同。密度预测通常是一个回归问题容量预测更接近带约束的分类或回归问题流量预测则是典型的时间序列外推问题。如果项目需求只写了“预测空域”首先要和业务确认输出口径否则后端数据和标签设计都会偏离。基础模型在这个环节的价值不是“一个模型解决所有空域问题”而是先学习空域状态的结构化表示再根据不同下游任务进行微调。比如同一个时空编码器可以同时支持密度预测和扇区容量评估。不过这个能力建立在任务定义清晰、评估指标可靠、数据切分正确的前提上。1.2 Foundation Model 在空域预测中的角色Foundation Model 的通用套路是先在大规模数据上预训练得到对领域结构的通用表示再微调到具体任务。在自然语言和视觉领域这种模式已经成熟。空域数据本质上也有很强的时空结构飞行目标不是静止出现的它的当前位置、速度、方向和趋势之间存在连续性天气影响会在空间上蔓延航班计划会带来周期性流量高峰。这些规律可以提前用预训练任务学习。目前并没有一个广泛开放、被行业公认的“空域基础模型”多数项目仍然是针对单一任务训练的专用模型。更现实的落地路径是用一个轻量 Transformer 模型针对“未来 30 分钟网格密度预测”这个任务做端到端训练同时保留“共享编码器 下游预测头”的结构。这样后续加入新任务时可以复用编码器做微调避免每个任务都从零训练。1.3 原型任务设定为了让文章可复现这里明确一个最小任务输入过去 12 个时间步的网格密度每步 5 分钟共 1 小时。输出未来 6 个时间步的网格密度共 30 分钟。空间范围16×16 的等距网格。评估方式MSE、RMSE、MAE以及预测热力图检查。这个任务看起来简单但已经覆盖了空域预测的核心难点空间局部性、时间趋势、输出多步序列。后面的代码都围绕这个任务展开。2. 准备数据和依赖让“空域”变成可学习的张量2.1 环境准备与依赖对齐建议使用独立虚拟环境避免包版本污染。项目目录可以命名为airspace_foundation_modelmkdir airspace_foundation_model cd airspace_foundation_model python -m venv venv source venv/bin/activateWindows 环境使用venv\Scripts\activate。激活后先升级 pip再安装依赖pip install --upgrade pip pip install numpy pandas scikit-learn torch matplotlib如果只需要 CPU 训练可以从 PyTorch 官方 CPU 源安装减少安装体积pip install torch --index-url https://download.pytorch.org/whl/cpu这里要注意PyTorch 安装源和版本会影响后续代码能否运行。生产或多人协作时建议把依赖锁定到具体版本pip freeze requirements.txt如果别人的环境跑不出你的指标先检查requirements.txt是否一致这是空域预测项目里非常常见的复现问题。2.2 空间网格和时间窗口设计空域数据不是天然图片经纬度坐标需要映射到网格或扇区。网格的选择直接影响模型复杂度网格过大多个航路点被合并空间细节丢失预测结果偏“糊”。网格过小大量网格长期为空数据稀疏训练效率低显存消耗高。原型先用 16×16 网格跑通是合理的。后续根据目标区域大小和数据密度调整到 32×32 或更大。如果使用经纬度网格在高纬度地区经度距离会明显变短实际项目建议使用等距投影或者直接使用扇区、航路点聚合后的数据。时间窗口同样影响任务。输入窗口太长模型会花费更多参数学习冗余历史太短则难以捕捉趋势。示例选择 12 个时间步输入、6 个时间步输出每个时间步 5 分钟正好覆盖 1 小时输入到 30 分钟预测的常见业务周期。窗口参数应该在实验阶段固定不要频繁改动训练集否则很难对比模型版本。2.3 生成最小仿真数据集在拿到真实 ADS-B、雷达或航班计划数据之前可以用高斯密度源模拟多个“飞行热点”移动。这个仿真数据的意义是验证建模链路不代表真实空域分布。生成函数如下import numpy as np def _gauss(grid_h, grid_w, cx, cy, intensity): yy, xx np.mgrid[0:grid_h, 0:grid_w] return (intensity * np.exp(-((xx - cx) ** 2 (yy - cy) ** 2) / 8.0)).astype(np.float32) def generate_airspace_samples(num_samples1200, grid_h16, grid_w16, seq_len12, pred_len6, seed42): rng np.random.default_rng(seed) X np.zeros((num_samples, seq_len, grid_h, grid_w), dtypenp.float32) Y np.zeros((num_samples, pred_len, grid_h, grid_w), dtypenp.float32) for i in range(num_samples): cx rng.uniform(3, grid_w - 3) cy rng.uniform(3, grid_h - 3) vx rng.uniform(-0.15, 0.15) vy rng.uniform(-0.15, 0.15) intensity rng.uniform(0.5, 2.0) for t in range(seq_len): X[i, t] _gauss(grid_h, grid_w, cx vx * t, cy vy * t, intensity) X[i, t] X[i, t] rng.normal(0, 0.02, size(grid_h, grid_w)).astype(np.float32) for t in range(pred_len): Y[i, t] _gauss(grid_h, grid_w, cx vx * (seq_len t), cy vy * (seq_len t), intensity) return X, Y生成的数据形状为(样本数, 时间步, 网格高, 网格宽)。每个样本是一个缓慢移动的密度源叠加轻微噪声用来模拟飞行目标的连续运动和位置不确定性。真实项目中的数据源比这复杂得多航班计划给出未来需求气象雷达给出天气影响ADS-B 或雷达给出当前位置和轨迹。但这个仿真数据足以让模型链路先跑起来后续替换数据源时只需要保持张量形状一致。2.4 数据标准化与时间序列切分神经网络对输入的尺度比较敏感尤其是 MSE 损失。建议先做标准化X, Y generate_airspace_samples() X_flat X.reshape(-1) Y_flat Y.reshape(-1) x_mean, x_std X_flat.mean(), X_flat.std() 1e-6 y_mean, y_std Y_flat.mean(), Y_flat.std() 1e-6 X_norm (X - x_mean) / x_std Y_norm (Y - y_mean) / y_std标准化参数必须在训练集上计算再应用到验证集和测试集避免验证集信息泄漏到训练过程中。切分方式同样是关键。真实空域数据必须按时间顺序切分不能随机打乱。如果随机把相邻时间片段分到训练集和验证集模型相当于在验证集里看到了“未来”的高度相似样本上线后指标会明显变差。import torch from torch.utils.data import TensorDataset, DataLoader, Subset dataset TensorDataset(torch.from_numpy(X_norm), torch.from_numpy(Y_norm)) train_size int(0.8 * len(dataset)) train_ds Subset(dataset, range(train_size)) val_ds Subset(dataset, range(train_size, len(dataset))) train_loader DataLoader(train_ds, batch_size32, shuffleTrue) val_loader DataLoader(val_ds, batch_size32, shuffleFalse)仿真数据样本之间可以视为独立但这里仍然使用顺序切分目的是让代码风格更贴近真实空域数据的处理习惯。3. 设计轻量时空基础模型先用 Transformer 验证时空关联3.1 模型输入输出定义模型结构按下面这条链路设计输入形状[B, T, H, W]其中 B 是 batch sizeT 是输入时间步H 和 W 是网格高宽。在通道维度补一维变成[B, T, 1, H, W]。用一个轻量卷积网络把每个时间步的网格图映射成向量。加上时间位置编码输入 TransformerEncoder。取最后一个时间步的编码结果通过 MLP 输出未来 6 个时间步的网格预测。这个设计不是最复杂的但足够验证时空关联是否被模型学到。它假设最后一个时刻的隐藏状态包含足够多的预测信息TransformerEncoder 的注意力机制负责捕捉历史时间步之间的依赖关系。3.2 卷积嵌入与位置编码先写卷积嵌入模块import torch import torch.nn as nn import math class ConvEmbedding(nn.Module): def __init__(self, input_channels, embed_dim): super().__init__() self.conv nn.Sequential( nn.Conv2d(input_channels, 32, kernel_size3, padding1), nn.GELU(), nn.Conv2d(32, embed_dim, kernel_size3, padding1), nn.GELU(), ) self.norm nn.LayerNorm(embed_dim) def forward(self, x): # x: [B, T, C, H, W] B, T, C, H, W x.shape x x.reshape(B * T, C, H, W) x self.conv(x) x x.mean(dim(2, 3)) x x.reshape(B, T, -1) return self.norm(x)这里的卷积层负责提取空间局部特征。mean(dim(2, 3))是全局平均池化把整张热力图压缩成一个向量适合关注整体密度变化的任务。代价是损失空间细节如果后续要做精细化网格预测需要换成输出空间位置 token 的编码方式而不是全局池化。Transformer 本身不感知顺序所以还要加入时间位置编码def get_position_encoding(seq_len, d_model): pe torch.zeros(seq_len, d_model) position torch.arange(0, seq_len, dtypetorch.float).unsqueeze(1) div_term torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model)) pe[:, 0::2] torch.sin(position * div_term) pe[:, 1::2] torch.cos(position * div_term) return pe.unsqueeze(0)位置编码让模型知道每个编码向量对应哪个时间步否则模型只能依赖输入分布来区分时间很难学习“过去第 3 步到第 5 步的变化趋势”这类时间模式。3.3 Transformer 编码器与预测头把模块组合成完整模型class AirspaceFoundationModel(nn.Module): def __init__(self, input_channels1, embed_dim128, nhead4, num_layers3, max_seq_len12, pred_len6, grid_h16, grid_w16): super().__init__() self.grid_h grid_h self.grid_w grid_w self.pred_len pred_len self.embed ConvEmbedding(input_channels, embed_dim) self.register_buffer(pos_embed, get_position_encoding(max_seq_len, embed_dim)) encoder_layer nn.TransformerEncoderLayer( d_modelembed_dim, nheadnhead, dim_feedforward256, batch_firstTrue, ) self.transformer nn.TransformerEncoder(encoder_layer, num_layersnum_layers) self.head nn.Sequential( nn.Linear(embed_dim, 256), nn.GELU(), nn.Dropout(0.1), nn.Linear(256, pred_len * grid_h * grid_w), ) def forward(self, x): # x: [B, T, H, W] B, T, H, W x.shape x x.unsqueeze(2) # [B, T, 1, H, W] src self.embed(x) src src self.pos_embed[:, :T, :] memory self.transformer(src) last memory[:, -1, :] out self.head(last) return out.reshape(B, self.pred_len, self.grid_h, self.grid_w)这里有一个细节pos_embed[:, :T, :]在输入序列长度小于max_seq_len时可以正常工作但如果序列长度超过max_seq_len会越界。生产环境建议改成动态位置编码或者对输入长度做硬限制。关键超参数总结参数含义示例值调大影响调小影响embed_dim编码向量维度128表达能力强显存增加表达受限训练更快nhead注意力头数4增强多头关系显存增加关系建模能力下降num_layersTransformer 层数3能捕捉更复杂时序易过拟合