在实际的空气质量预测项目中模型性能往往不是只由网络层数决定而是由数据的时间尺度、上下文信息和预测目标之间的匹配程度决定。AirFlow 这个模型方向的核心也在这里它不追求用一个固定分辨率处理所有输入而是把“上下文保留”和“多速率状态建模”作为设计主线专门解决污染物浓度在小时级、日级、周级不同频率上的演化规律。本文从工程落地视角拆解这个模型思路给出数据准备、模型结构、训练评估、常见排错和生产部署建议适合正在做时序预测、环境数据分析或城市计算相关项目的开发者参考。1. 从同名歧义开始AirFlow 模型解决的不是调度问题1.1 为什么空气质量预测需要专门的建模方法空气质量预测在形式上是一个多变量时序预测任务输入历史一段时间内的污染物浓度和气象数据输出未来若干小时的 PM2.5、PM10、NO2、O3 等浓度值。但它比普通时序预测更复杂的地方在于污染物浓度受到多个物理过程的共同作用。局地机动车排放、工业活动、建筑扬尘会在小时尺度上造成浓度突变气象系统风速、风向、边界层高度会在数小时到数天内改变污染物的扩散条件季节性供暖、降水模式和大气环流则带来更缓慢的长期趋势。如果只用固定采样间隔的数据比如统一重采样到小时模型很难同时捕捉分钟级的峰值突变和天气系统带来的日际变化。AirFlow 模型的出发点正是保留多个时间分辨率的输入让不同频率的状态信息各司其职。1.2 上下文保留和多速率状态建模分别指什么上下文保留Context Preserving关注的是模型在处理长序列时如何让历史信息不丢失。常见的 RNN/LSTM 模型虽然能记住过去但记忆会随着时间步增长而衰减遇到极端污染事件或者天气转折时模型容易把重点放在最近的输入上忽略几天前已经出现的累积趋势。上下文保留机制通常通过门控、注意力或额外记忆单元实现让模型在生成当前状态时有选择地保留长期上下文。多速率状态建模Multi-Rate State Modeling关注的是输入序列本身的采样频率。传统做法把所有特征重采样到同一频率例如全部转成小时数据这样实现简单但会牺牲高频细节。AirFlow 的做法是并行维护多条时间分支分钟级分支捕捉局地突变小时级分支建模扩散和气象过程日级或周级分支刻画季节背景最后在设计好的融合模块中合并这些状态。1.3 不要把方法名和工作流调度平台搞混搜索“AirFlow”时大量结果会指向 Apache AirFlow那是开源的工作流调度平台用于编排数据处理任务、定时触发 ETL 流程和空气质量预测没有直接关系。AirFlow 作为学术模型名称主要出现在论文、技术报告和时序预测项目复现里。如果目标是学习调度平台需要查找 Apache AirFlow 官方文档并重点确认版本差异如果目标是做空气质量预测本文后续内容围绕模型方法展开不涉及调度平台安装部署。理解这一点很重要因为后续的代码和配置都基于 PyTorch 和数据处理工具不是基于任务调度平台。2. 问题定义与数据准备先想清楚要预测什么2.1 任务形式化和预测目标在动手写模型之前先把任务写成数学形式。假设某个监测站点在时刻 t 的观测向量为 X_t包含 D 个特征例如 PM2.5、PM10、SO2、NO2、CO、O3、温度、湿度、风速、风向、气压。给定过去 L 个时间步的历史观测X_{t-L1}, X_{t-L2}, ..., X_t目标是预测未来 H 个时间步的污染物浓度Y_{t1}, Y_{t2}, ..., Y_{tH}最常见的设定是输入过去 72 小时数据预测未来 24 小时 PM2.5 或 AQI。这里的 H 决定模型是单步预测还是多步预测。多步预测比单步更难因为误差会逐步累积尤其在边界层变化剧烈的清晨和傍晚时段。如果把任务扩展到多个站点则 X_t 变成站点集合上的图信号矩阵需要额外引入站点间距离、风向路径等信息。AirFlow 的思路在单站点上先适用多站点场景可以看作其在空间维度的扩展。2.2 数据来源与字段设计空气质量预测常用的公开数据包括中国环境监测总站发布的全国城市空气质量实时数据、一些高校和科研机构整理的城市污染物数据集以及 Kaggle、UCI 等平台上经过清洗的历史数据。实际使用前要确认数据的时间范围、更新频率、授权方式和字段缺失情况。常见字段可以整理成表便于后续编写预处理代码时对照字段类型示例说明timestampdatetime2024-01-15 08:00:00观测时间注意时区统一station_idstring1001A站点编码多站点场景关键pm25float78.5PM2.5 浓度单位微克/立方米pm10float120.3PM10 浓度no2float56.2二氧化氮浓度o3float30.1臭氧浓度temperaturefloat12.6气温单位摄氏度humidityfloat45.0相对湿度单位百分比wind_speedfloat2.3风速单位米/秒wind_directionfloat180.0风向角度0-360 度pressurefloat1023.5气压单位 hPa需要注意的是风向是环形变量直接作为数值特征会让 0 度和 360 度看起来差异很大实际工程中建议拆成 sin 和 cos 两个分量或者按 16 方位分桶编码。2.3 多速率数据对齐不要把高频信息直接磨平多速率建模的第一步不是把所有数据塞进同一个 DataFrame而是按业务需求生成多个频率的序列。比如原始监测数据是 5 分钟一条气象数据是 1 小时一条想同时保留小时级波动和日级趋势可以生成三条分支5 分钟分支、1 小时分支、1 天分支。这里推荐用 pandas 的 resample 完成import pandas as pd def build_multirate_series(df, value_colpm25, rulesNone): if rules is None: rules { high: 5min, mid: 1h, low: 1D, } df df.copy() df[timestamp] pd.to_datetime(df[timestamp]) df df.set_index(timestamp).sort_index() series {} for name, rule in rules.items(): # mean 用于整体浓度max 可以单独保留峰值分支 series[name] df[value_col].resample(rule).mean() return series代码块的思路是同一列污染物按不同窗口聚合得到高频、中频、低频三条曲线。实际项目中高频分支建议额外保留一个“最大值”序列因为空气质量预测对污染峰值更敏感均值会把短时高浓度事件磨平。注意重采样前必须检查原始时间戳是否连续、是否存在重复时间。重复时间戳会直接导致 resample 报错或者统计出错误结果。高频和低频分支的长度可能相差很大例如 5 分钟分支一天 288 个点日分支一天只有 1 个点。模型不能直接把这些序列拼接起来需要按窗口切分后再输入不同的编码器。3. 核心模型设计多分支编码与上下文门控3.1 多速率时间编码器结构AirFlow 的模型结构可以理解为一个“多分支编码器 上下文融合模块 输出层”。每个频率分支独立处理对应分辨率的序列然后把各自得到的隐状态融合起来输出未来预测值。下面是一个简化的 PyTorch 实现只用于说明思路不力求复现论文中的全部细节import torch import torch.nn as nn class RateEncoder(nn.Module): def __init__(self, input_size, hidden_size, num_layers1): super().__init__() self.gru nn.GRU( input_sizeinput_size, hidden_sizehidden_size, num_layersnum_layers, batch_firstTrue, ) def forward(self, x): # x: (batch, time, feature) output, hidden self.gru(x) return output, hidden # hidden: (layers, batch, hidden) class AirFlowStyleModel(nn.Module): def __init__(self, input_sizes, hidden_size, horizon): super().__init__() # input_sizes 是各频率分支的特征维度例如 [6, 6, 6] self.encoders nn.ModuleList([ RateEncoder(size, hidden_size) for size in input_sizes ]) self.context_fc nn.Linear(hidden_size * len(input_sizes), hidden_size) self.predictor nn.Linear(hidden_size, horizon) def forward(self, multi_rate_inputs): hiddens [] for x, encoder in zip(multi_rate_inputs, self.encoders): _, hidden encoder(x) # 取最后一层最后一刻的隐状态 hiddens.append(hidden[-1]) context torch.cat(hiddens, dim-1) context torch.tanh(self.context_fc(context)) out self.predictor(context) return out这段代码里每个分支使用独立的 GRU特征维度由 input_sizes 配置。高、中、低频率分支不共享参数因为 5 分钟分辨率上的波动模式和 1 天分辨率上的季节模式本质不同共享参数反而会引入尺度冲突。3.2 上下文保留机制让长期记忆参与融合仅仅把每条分支最后一个隐状态拼接在一起还不足以体现“上下文保留”。GRU 的最后一个隐状态仍然可能把几天前的重要信息压缩掉。更合理的方式是在融合前引入一个上下文门控模块让历史记忆和当前窗口信息做加权组合。class ContextGate(nn.Module): def __init__(self, hidden_size): super().__init__() self.gate nn.Linear(hidden_size * 2, hidden_size) def forward(self, current_state, long_memory): # current_state: 近期模式编码 # long_memory: 长期上下文向量 g torch.sigmoid(self.gate(torch.cat([current_state, long_memory], dim-1))) return g * current_state (1 - g) * long_memory这里的 long_memory 可以来自一个额外的慢速编码器比如对过去 30 天日序列做注意力池化得到。门控的值越接近 1表示模型越信任当前窗口信息越接近 0表示越依赖长期上下文。这种机制在天气状态发生转折时尤其重要当污染物浓度已经明显升高时模型需要判断这是短期局地排放还是长期静稳天气累积造成的不能单纯根据最近几小时数据做决策。除了门控注意力机制也是非常直接的上下文保留方式。对 encoder 输出的整条序列做加权求和模型可以自动学习哪段时间最重要。实际项目中可以两者结合门控用来控制旧记忆和新信息的比例注意力用来挑选历史时间步中的关键片段。3.3 损失函数与训练逻辑空气质量预测本质是回归任务最常见损失函数是 MSE 和 MAE。MSE 对大误差惩罚更重会推动模型更关注浓度峰值但也容易被个别异常观测主导MAE 更稳定但在梯度更新时对误差正负方向处理不同。很多项目会选择 Huber Loss它在小误差时接近 MSE在大误差时接近 MAE兼顾收敛速度与鲁棒性criterion nn.SmoothL1Loss()如果业务更关注高污染时段可以对预测误差按真实浓度加权。例如真实浓度超过 150 微克/立方米时损失权重放大 1.5 到 2 倍让模型把更多能力放在高风险区间。训练循环没有特殊之处但时间序列预测要注意数据划分方式。不能用随机打乱训练集和测试集必须按时间顺序切分否则未来信息会泄漏到训练集中验证指标会虚高。optimizer torch.optim.Adam(model.parameters(), lr1e-3) scheduler torch.optim.lr_scheduler.ReduceLROnPlateau( optimizer, modemin, factor0.5, patience5 ) for epoch in range(max_epochs): model.train() for batch in train_loader: optimizer.zero_grad() pred model(batch[inputs]) loss criterion(pred, batch[targets]) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0) optimizer.step() val_loss evaluate(model, val_loader) scheduler.step(val_loss) if early_stopping.should_stop(val_loss): break训练中比较关键的细节是梯度裁剪。多分支模型合并时不同频率分支的梯度尺度可能差异很大如果不加 clipping低频分支偶尔会把高频分支的权重更新方向带偏。4. 从零复现最小案例数据预处理、训练与验证4.1 环境依赖与项目结构复现这个方向不需要太重的依赖建议使用以下环境依赖版本建议用途Python3.9 及以上基础运行环境PyTorch2.0 及以上模型搭建与训练pandas1.5 及以上数据处理与重采样numpy1.24 及以上数值计算scikit-learn1.2 及以上数据切分与评估指标matplotlib3.6 及以上预测曲线可视化安装时使用 pip 即可实际项目要以自己的系统环境为准pip install torch pandas numpy scikit-learn matplotlib项目结构建议按下面这种方式组织方便后续扩展airflow_forecast/ ├── data/ │ ├── raw/ # 原始监测数据 │ └── processed/ # 预处理后的多速率序列 ├── src/ │ ├── data_preprocess.py # 多速率重采样与缺失值处理 │ ├── model.py # 模型结构定义 │ ├── train.py # 训练与验证脚本 │ └── evaluate.py # 指标计算与可视化 ├── configs/ │ └── default.yaml # 超参数配置 └── notebooks/ └── explore.ipynb # 数据探索4.2 数据预处理归一化、缺失值与时间窗口切分预处理最容易犯的错误是“先整体归一化再划分数据集”。正确做法是只用训练集拟合归一化参数再用它转换验证集和测试集。下面的代码演示了最小处理流程import numpy as np import pandas as pd from sklearn.preprocessing import StandardScaler def preprocess_for_training(df, feature_cols, target_colpm25): # 1. 保证时间排序 df df.sort_values(timestamp).reset_index(dropTrue) # 2. 数值特征填充 df[feature_cols] df[feature_cols].fillna(methodffill) # 3. 只在训练数据上拟合归一化参数 train_size int(len(df) * 0.7) train_df df.iloc[:train_size] test_df df.iloc[train_size:] scaler StandardScaler() train_x scaler.fit_transform(train_df[feature_cols]) test_x scaler.transform(test_df[feature_cols]) train_y train_df[target_col].values test_y test_df[target_col].values return train_x, test_x, train_y, test_y, scaler这段代码的意图是先用前向填充处理缺失值再按时间切分最后在训练集上 fit scaler。实际项目中“前向填充”不一定是最好的策略如果缺失时间较长还要结合线性插值或基于相似日的填充但要避免直接引入未来数据。窗口切分时要按固定步长滑动生成样本同时保证目标值不会越过时间边界造成泄漏def make_windows(x, y, input_len, horizon, step1): xs, ys [], [] for start in range(0, len(x) - input_len - horizon 1, step): end start input_len xs.append(x[start:end]) ys.append(y[end:end horizon]) return np.array(xs), np.array(ys)这里 input_len 是历史窗口长度horizon 是预测长度step 是滑动窗口步长。step 越大生成的样本越少训练越快但数据利用率越低。4.3 训练脚本与验证流程多速率输入在 DataLoader 中需要同时返回多个频率的数组。如果同时使用 5 分钟、1 小时、1 天三条分支输入窗口对应的时间长度必须一致比如都以“过去 3 天”为准那么三条分支的形状分别是 (batch, 864, feature)、(batch, 72, feature)、(batch, 3, feature)。模型训练和验证的伪代码可以这样组织def train_one_epoch(model, loader, optimizer, criterion): model.train() total_loss 0.0 for batch in loader: high batch[high] mid batch[mid] low batch[low] target batch[target] optimizer.zero_grad() pred model([high, mid, low]) loss criterion(pred, target) loss.backward() optimizer.step() total_loss loss.item() return total_loss / len(loader)验证阶段除了计算损失还要保存可视化结果把测试集上的预测序列和真实序列画在一起。这一步很关键因为 RMSE 只能看到一个数字而曲线图能直观展示模型是否存在系统性滞后比如预测曲线比真实曲线晚 6 小时这种问题在数值指标上很难察觉。5. 参数选型与评估设计5.1 核心参数速查表多速率模型的控制参数比单模型多下面是几个最重要的参数及其影响参数推荐初始值调大影响调小影响注意事项高频窗口长度72 小时上下文更长训练成本上升只保留近期信息可能丢失缓变过程要保证各频率时间范围一致分支数量3信息维度更多但融合变难实现简单但频率信息不完整不是越多越好先做 2 到 3 条hidden_size64拟合能力更强容易过拟合模型容量不足长期趋势学不住结合样本量选择学习率1e-3收敛快可能震荡训练慢适合精细调参使用 ReduceLROnPlateau 自动调节批大小64梯度更稳定内存占用更大更新频繁容易抖动序列较长时应减小批大小dropout0.2正则更强训练变慢过拟合风险高多分支融合层尤其建议加 dropout预测步长24 小时业务价值高误差累积明显误差更小但短期预测实用性有限建议同时训练多个 horizon 头参数调节不能只盯训练损失。多分支模型容易在训练集上表现很好验证集却出现大的偏差所以要同时观察训练和验证曲线差距过大说明过拟合应该增加 dropout、减少分支数量或增大数据量。5.2 评估指标选择常用的回归指标都能用但要结合业务场景解读RMSE对大幅误差敏感适合评估模型在高浓度事件上的表现。单位与浓度一致便于业务理解。MAE反映平均绝对误差比 RMSE 更稳健不受个别离群点主导。MAPE百分数误差直观但 PM2.5 接近 0 时会变得极大且不稳定冬春清洁天气时慎用。R²反映模型对目标方差的解释程度但不能反映预测是否系统偏小。如果预测目标是 AQI 等级或者是否超标还需要计算分类准确率、召回率、误报率。例如预测未来 24 小时是否会出现 PM2.5 超过 150 微克/立方米的高污染事件即使浓度数值有误差只要等级判断正确对预警也有价值。评估时建议按浓度区间分组统计而不是只看全量指标。把测试集按照真实 PM2.5 浓度分成“低于 75、75 到 150、高于 150”三组分别计算 RMSE可以快速发现模型是否只在低浓度区间表现好高浓度区间失效。5.3 对照实验设计要验证 AirFlow 的多速率和上下文保留模块确实有效不能只跑一个模型就说效果不错。建议设计以下对照实验实验输入模型目的基线小时级单分支LSTM/GRU验证多速率的增益单速率强化小时级单分支更大隐藏层相同结构排除参数规模导致的效果差异多速率无上下文门控多分支无 ContextGate验证上下文保留的作用AirFlow 风格多分支完整结构综合结果实验过程中必须固定随机种子、数据划分和归一化参数否则无法判断效果差异来自模型还是数据。每一组实验至少跑 3 个随机种子取平均值和标准差结论才可靠。6. 常见问题排查数据缺失、粒度错位与预测漂移6.1 高频问题对照表实际复现中经常出现问题可以先按下面表格快速定位问题现象常见原因检查方式处理建议训练 loss 出现 NaN数据缺失值未处理或归一化参数为 0打印输入数据的 min/max检查是否包含 NaN删除或插值缺失时间戳归一化前检查方差预测曲线整体滞后数小时输入窗口和预测窗口的时间对应关系写错画出一次具体样本的输入、真实目标、预测目标检查数据切分确认预测目标从 t1 开始多分支融合后效果不如单分支频率分支特征没有对齐或融合层参数过多分别单独训练每个分支并记录指标先跑单分支再逐步叠加分支观察增益高浓度时段预测偏低MSE 把常规时段主导峰值样本占比低按浓度区间统计 RMSE使用峰值加权损失或 Huber Loss模型在验证集上波动大数据划分随机化或样本量太小固定随机种子重跑改用时间顺序切分增加交叉验证6.2 三个典型问题排查链路第一类是数据缺失。现象是训练时某些高频分支的长度不固定或者测试时出现 NaN。先检查原始数据的每分钟记录数量分布确认是否存在停站、断网等长时间缺失。处理方式不是直接删除整段时间而是对缺失段打标记并在预处理阶段用插值或“前值”填充。更稳妥的方式是把“缺失时长”作为一个额外特征传入模型让模型自己学习缺失时间对预测的影响。第二类是粒度错位。多速率模型要保证每个分支覆盖的物理时间范围一致。比如想用过去 3 天预测未来 24 小时高频 5 分钟分支应截取最近 864 个点小时分支应截取最近 72 个点日分支应截取最近 3 个点。如果小时分支错取成最近 72 个点但实际是 24 小时前的数据模型会从错误的起点开始学习结果通常表现为预测滞后或周期性抖动。第三类是长期预测漂移。多步预测误差累积是常见现象尤其在 horizon 较长时。可以尝试三种策略一是直接多步输出让模型一次输出未来 24 小时二是自回归迭代把上一时刻预测值当作下一步输入但误差累积明显三是先训练单步模型推理时采用教师强制和预测输入混合的策略降低漂移。实践中经常发现直接多步输出反而更适合长期预测因为模型不需要反复把噪声预测值当成真实输入。注意验证指标只显示“平均 RMSE 下降”是不够的还要看预测曲线是否出现系统性时间滞后。如果预测曲线形状正确但整体向右平移说明模型学到的是延迟复制而不是真正的趋势预测。7. 生产环境落地要点与扩展方向7.1 从离线预测到在线推理的改造离线实验跑通后生产环境会引入新的问题数据实时性、推理延迟和模型更新频率。在线预测场景下模型一般以小时或分钟频率被调用输入是最近一段时间的监测数据输出是未来 24 小时的浓度曲线。在线推理需要把数据预处理和模型加载做成一个服务。建议在服务启动时一次性加载模型权重和归一化参数并把已经计算好的特征缓存到内存中避免每次请求都从数据库重新拉取并重采样全部历史数据。重采样计算量虽然不大但频繁执行会造成不必要的延迟和 IO 压力。接口输出还需要设定降级策略。如果上游数据中断超过一定时间模型给出的输入窗口不完整此时返回缓存的上一次预测结果或直接返回“数据不可用”而不是输出一个看似正常但没有依据的预测值。7.2 模型维护与监控生产模型不是训练完就结束需要持续监控预测偏差。每天把真实浓度和模型预测值对比计算滚动偏差如果发现系统偏差持续超过阈值就要触发重训流程。重训前先排查数据质量因为站点仪器更换、传感器漂移、时间字段格式变化都会让模型输入分布发生偏移。模型版本管理也很重要。每次重训后要记录数据版本、特征版本、模型版本和评估指标并保留上一个版本的权重便于回滚。归一化参数一旦重新拟合必须和对应模型打包发布不能单独替换。空气质量预测模型上线前还要考虑业务指标和模型指标的关系。例如预警阈值是否依赖预测曲线的最大值而不是平均值。如果业务上最关心是否出现重度污染模型评估就要重点关注高浓度时段的误差和高污染事件的召回率。7.3 扩展方向AirFlow 的多速率和上下文保留思路可以继续向几个方向扩展。第一引入气象预报数据。历史气象数据用于训练未来气象预报数据可以加入 decoder让模型知道未来风速、降水和风向的变化这是提升中长期预测效果的重要路径。第二加入站点间空间关系。多个监测站点之间存在空间相关性下游站点的污染往往来自上游城市传输。可以引入图神经网络把站点邻接关系和风向路径建模成图结构让每个站点的状态更新时考虑邻近站点的信息。第三结合物理约束。纯数据驱动模型容易预测出不符合物理规律的浓度值比如在持续强降雨时段预测 PM2.5 大幅上升。可以在损失函数中加入物理正则项或者在模型结构中加入排放、扩散等物理过程的简化模拟层提升预测合理性。第四不确定性量化。空气质量预测对决策非常重要仅输出一个点预测不够。可以使用深度集成、MC Dropout 或分位数回归输出预测区间给决策层提供风险信息。实际复现时不建议一开始就追求完整论文结构。先从单一站点、两个频率分支入手把数据预处理、模型训练、评估和可视化链路跑通再逐步加入上下文门控、更多频率分支和站点空间关系。这个过程中数据分布、时间对齐和评估粒度的问题往往比模型结构更影响最终效果值得优先投入精力。