PINN+LSTM混合模型:物理约束提升温度场预测泛化能力
2026/8/30 5:41:13 网站建设 项目流程

之前做区域级温度场预测项目时,最头疼的问题是:纯数据驱动的 LSTM 模型在历史工况下拟合得不错,可一旦把时间轴向前推进、热源位置发生偏移,预测结果就开始发散,甚至出现违背热传导常识的“温度断层”。后面我把物理信息神经网络(PINN)引入,让物理方程和 LSTM 的时序特征在同一个模型里协同工作,预测精度和跨工况泛化能力都有了明显提升。

这篇文章会把这套PINN + LSTM 混合建模的完整思路、核心原理、可运行代码和工程落地经验整理出来。我们会用到区域温度场的二维热扩散问题作为演示场景,用 PyTorch 从零搭建模型,最终对比纯 LSTM 与 PINN+LSTM 在 RMSE、R² 以及跨工况外推上的差异。无论你是正在做时间序列预测,还是想了解物理约束神经网络如何落地,这篇文章都能给你一份系统性参考。

1. 背景与核心概念

1.1 纯数据驱动模型为什么不够用

以区域温度场预测为例,传感器或网格节点会按固定频率采集温度数据,这样每个空间点都形成一条时间序列。LSTM 天然适合学习这种序列中的长短期依赖,因此在常规测试集上往往表现很好。

但问题在于:LSTM 学到的本质上是“训练数据分布内的模式”。一旦工况发生改变,比如热源中心偏移、边界条件变化、设备老化导致导热系数漂移,模型输入就会落在训练分布之外。此时 LSTM 只能凭借历史统计相关性做外推,结果常常不符合物理规律,例如出现温度突变、局部过冲、负温度等异常。

这类问题的核心矛盾是:纯数据驱动模型没有把已知的物理规律作为先验知识注入网络。它只关注“输出与标签一致”,不关注“预测结果是否满足控制方程”。

1.2 PINN 是什么、解决什么问题

物理信息神经网络(Physics-Informed Neural Network,PINN)是一种把物理方程嵌入神经网络训练过程的建模方法。它的核心思想非常简单:除了常规的数据损失(Data Loss)之外,额外增加一项物理残差损失(Physics Residual Loss),迫使神经网络输出满足给定的偏微分方程,例如热扩散方程:

[ \frac{\partial u}{\partial t} = \alpha \left( \frac{\partial^2 u}{\partial x^2} + \frac{\partial^2 u}{\partial y^2} \right) ]

通俗地说,普通神经网络训练时只盯着“预测值 vs 真实值”,而 PINN 训练时还要盯着“预测函数的一阶导数、二阶导数是否满足方程”。网络不仅要拟合数据,还要成为一个满足物理规律的函数逼近器。

这种思路带来的直接收益有三点:

  • 小样本场景下,物理约束相当于额外正则项,降低过拟合风险。
  • 外推能力更强,即使测试工况超出训练数据分布,模型也会优先寻找“符合物理规律”的解。
  • 可解释性更好,模型输出可以被物理方程解释,而不是黑盒硬拟合。

1.3 为什么是 PINN + LSTM

PINN 擅长处理空间物理场,但传统 PINN 通常只输入坐标和时间,缺少历史时序特征。LSTM 擅长提取时间序列模式,但对空间网格之间的物理关联不敏感。二者结合正好互补:

  • LSTM 负责编码每个位置的历史温度变化趋势,提取“数据驱动”的时序特征。
  • PINN 分支负责接收空间坐标时间,输出满足热扩散方程的物理场预测。
  • 两者在融合层汇合,最终输出综合考虑时序信息与物理约束的预测结果。

在实际工程中,这种混合结构比单独使用 LSTM 更稳健,比单独使用 PINN 更容易拟合复杂实测数据,因此逐渐成为物理信息机器学习方向的一个标准基准。

2. PINN+LSTM 混合架构的核心原理

2.1 整体结构

整个模型可以拆成三个模块:

  1. LSTM 分支:输入某个空间点过去 T 步的温度序列,输出该点未来时刻的时序预测特征。
  2. PINN 分支:输入归一化后的时间 t 和空间坐标 (x, y),输出一个满足热扩散方程的物理场预测。
  3. 融合层:将 LSTM 输出、PINN 输出以及坐标时间信息拼接起来,通过一个小型 MLP 输出最终预测值。

训练时模型同时优化两部分损失:

  • 数据损失:最终预测值与真实温度之间的均方误差。
  • 物理损失:在随机采样点上计算 PINN 分支输出的时间导数与空间二阶导,检查是否满足热扩散方程。

2.2 物理残差的计算方式

PINN 最核心的操作是自动微分。PyTorch 的torch.autograd.grad可以直接计算网络输出对输入坐标的偏导数,从而构造物理残差:

[ r(t, x, y) = \frac{\partial u}{\partial t} - \alpha \left( \frac{\partial^2 u}{\partial x^2} + \frac{\partial^2 u}{\partial y^2} \right) ]

训练时希望这个残差在所有采样点上尽量接近 0。这样网络学到的函数不仅拟合了训练数据,还隐式满足热扩散方程。

2.3 损失函数设计

总损失函数为:

[ L = L_{data} + \lambda_{phys} L_{phys} ]

其中:

  • (L_{data} = \frac{1}{N}\sum (u_{pred} - u_{true})^2)
  • (L_{phys} = \frac{1}{M}\sum r(t_m, x_m, y_m)^2)
  • (\lambda_{phys}) 是物理损失权重,一般取 0.01~1.0,需要根据数据噪声和收敛情况调整。

这里的 (\lambda_{phys}) 不是越大越好。权重过大会让模型“只遵守物理规律”而忽略数据细节,权重过小则退化成普通 LSTM 模型。

2.4 评估指标:RMSE 与 R²

本文提到的两个核心指标解释如下:

  • RMSE(均方根误差):反映预测误差的绝对大小,单位与温度一致。RMSE 越小说明预测越准。计算公式为:

[ RMSE = \sqrt{\frac{1}{N}\sum_{i=1}^{N}(y_i - \hat{y}_i)^2} ]

  • R²(决定系数):反映模型对数据方差的解释程度,取值范围通常为 0~1。R²=0.99 表示模型解释了 99% 的数据波动,拟合能力已接近理想。

3. 环境准备与数据集说明

3.1 运行环境

本文代码基于 PyTorch 实现,推荐环境如下:

Python 3.9+ PyTorch 2.0+ NumPy Matplotlib scikit-learn

对应 requirements.txt 内容为:

torch>=2.0 numpy>=1.24 matplotlib>=3.7 scikit-learn>=1.2

如果你使用 GPU 训练,建议安装对应 CUDA 版本的 PyTorch;本文示例在 CPU 环境下也能在几分钟内跑通。

3.2 数据集怎么来

为了方便复现,我们采用二维热扩散方程的解析解来生成合成数据。解析解表达式为:

[ u(x,y,t) = \sum_{i} \frac{A_i}{1+4\alpha t} \exp\left( -\frac{(x-x_i)^2 + (y-y_i)^2}{4\alpha t + 2\sigma_i^2} \right) ]

这个表达式严格满足热扩散方程,用来生成训练数据不会引入数值误差。真实业务场景中,可以把这里替换为传感器实测数据或 CFD(计算流体力学)仿真数据。

3.3 数据集结构与划分

假设空间网格为 32×32,时间步数为 90。对每个网格点,取过去 12 步温度序列作为 LSTM 输入,预测未来一步的温度。

训练集取时间步 12~60 之间的样本,测试集取时间步 60~88 之间的样本。这种按时间切分的方式对应“跨工况外推”,能真实检验模型的泛化能力。

4. 完整实战:搭建 PINN+LSTM 区域预测模型

4.1 项目结构

pinn-lstm-demo/ ├── requirements.txt ├── data_gen.py # 生成热扩散合成数据 ├── dataset.py # 构造训练 / 测试样本 ├── model.py # 定义 LSTM 分支、PINN 分支、混合模型 ├── physics.py # 物理残差损失函数 ├── train.py # 训练主流程 └── evaluate.py # 评估与对比

4.2 步骤一:生成热场数据

# 文件路径:data_gen.py import numpy as np def generate_thermal_field(nx=32, ny=32, nt=90, alpha=0.02): """ 基于热扩散方程解析解生成二维温度场序列。 返回: u: shape (nt, nx, ny) """ x = np.linspace(0, 1, nx) y = np.linspace(0, 1, ny) X, Y = np.meshgrid(x, y, indexing="ij") # 初始热源参数: (中心x, 中心y, 峰值A, 展宽sigma) sources = [ (0.3, 0.4, 0.8, 0.10), (0.7, 0.5, 1.2, 0.08), (0.5, 0.8, 0.6, 0.12), ] times = np.linspace(0.01, 1.0, nt) u_seq = [] for t in times: u = np.zeros_like(X) for cx, cy, A, sigma in sources: u += A / (1 + 4 * alpha * t) * np.exp( -((X - cx) ** 2 + (Y - cy) ** 2) / (4 * alpha * t + 2 * sigma ** 2) ) u_seq.append(u) return np.stack(u_seq, axis=0)

这个函数生成 90 个时刻的温度场,每个温度场是 32×32 的二维矩阵。扩散系数 alpha 控制温度传播速度。

4.3 步骤二:构造训练样本

# 文件路径:dataset.py import numpy as np import torch from torch.utils.data import TensorDataset, DataLoader def build_samples(u, input_len=12, horizon=1): """ 将温度场序列转换为监督学习样本。 对每个空间点,用过去 input_len 步预测未来 horizon 步。 """ nt, nx, ny = u.shape xs = np.linspace(0, 1, nx) ys = np.linspace(0, 1, ny) hist_list, t_list, x_list, y_list, label_list = [], [], [], [], [] for i in range(nx): for j in range(ny): for start in range(input_len, nt - horizon): hist = u[start - input_len:start, i, j] # 历史序列 target = u[start + horizon, i, j] # 未来温度 pred_time = (start + horizon) / nt # 归一化时间 hist_list.append(hist) t_list.append(pred_time) x_list.append(xs[i]) y_list.append(ys[j]) label_list.append(target) return ( np.array(hist_list, dtype=np.float32), np.array(t_list, dtype=np.float32).reshape(-1, 1), np.array(x_list, dtype=np.float32).reshape(-1, 1), np.array(y_list, dtype=np.float32).reshape(-1, 1), np.array(label_list, dtype=np.float32).reshape(-1, 1), ) def make_loader(hist, t, x, y, label, batch_size=256, shuffle=True): dataset = TensorDataset( torch.tensor(hist).unsqueeze(-1), # (N, input_len, 1) torch.tensor(t), torch.tensor(x), torch.tensor(y), torch.tensor(label), ) return DataLoader(dataset, batch_size=batch_size, shuffle=shuffle)

注意这里把历史温度序列的维度补成(N, input_len, 1),因为 PyTorch 的 LSTM 默认输入维度是(batch, seq_len, input_size)

4.4 步骤三:定义 PINN+LSTM 模型

# 文件路径:model.py import torch import torch.nn as nn class LSTMBranch(nn.Module): """LSTM 时序分支:提取历史温度序列特征。""" def __init__(self, input_size=1, hidden_size=64, num_layers=2): super().__init__() self.lstm = nn.LSTM( input_size=input_size, hidden_size=hidden_size, num_layers=num_layers, batch_first=True, ) self.head = nn.Sequential( nn.Linear(hidden_size, 32), nn.ReLU(), nn.Linear(32, 1), ) def forward(self, hist): # hist: (B, T, 1) _, (h_n, _) = self.lstm(hist) feat = h_n[-1] # 取最后一层隐状态 return self.head(feat) class PINNBranch(nn.Module): """物理信息分支:输入 (t, x, y),输出物理场预测值。""" def __init__(self, hidden_dim=128): super().__init__() self.net = nn.Sequential( nn.Linear(3, hidden_dim), nn.Tanh(), nn.Linear(hidden_dim, hidden_dim), nn.Tanh(), nn.Linear(hidden_dim, hidden_dim), nn.Tanh(), nn.Linear(hidden_dim, 64), nn.Tanh(), nn.Linear(64, 1), ) def forward(self, t, x, y): inp = torch.cat([t, x, y], dim=1) return self.net(inp) class PINNLSTMModel(nn.Module): """PINN + LSTM 混合模型。""" def __init__(self, lstm_hidden=64, pinn_hidden=128): super().__init__() self.lstm_branch = LSTMBranch(hidden_size=lstm_hidden) self.pinn_branch = PINNBranch(hidden_dim=pinn_hidden) self.fusion = nn.Sequential( nn.Linear(5, 32), # lstm_out, pinn_out, t, x, y nn.ReLU(), nn.Linear(32, 1), ) def forward(self, hist, t, x, y): out_lstm = self.lstm_branch(hist) out_pinn = self.pinn_branch(t, x, y) fuse_in = torch.cat([out_lstm, out_pinn, t, x, y], dim=1) return self.fusion(fuse_in), out_pinn, out_lstm class LSTMOnlyModel(nn.Module): """纯 LSTM 基线模型:用于对比实验。""" def __init__(self, hidden_size=64): super().__init__() self.lstm = nn.LSTM( input_size=1, hidden_size=hidden_size, num_layers=2, batch_first=True, ) self.head = nn.Sequential( nn.Linear(hidden_size, 32), nn.ReLU(), nn.Linear(32, 1), ) def forward(self, hist): _, (h_n, _) = self.lstm(hist) return self.head(h_n[-1])

PINN 分支采用Tanh激活函数而不是ReLU,因为物理损失需要计算二阶导数,Tanh全场光滑且二次可微,ReLU在零点不可导,容易导致物理残差震荡。

4.5 步骤四:计算物理残差

# 文件路径:physics.py import torch def physics_residual_loss(model, t, x, y, alpha=0.02): """ 计算热扩散方程残差损失: L_phys = mean( (du/dt - alpha * (d²u/dx² + d²u/dy²))² ) """ t.requires_grad_(True) x.requires_grad_(True) y.requires_grad_(True) u = model.pinn_branch(t, x, y) # 一阶时间导数 du_dt = torch.autograd.grad( u, t, grad_outputs=torch.ones_like(u), create_graph=True, retain_graph=True, )[0] # 一阶空间导数 du_dx = torch.autograd.grad( u, x, grad_outputs=torch.ones_like(u), create_graph=True, retain_graph=True, )[0] du_dy = torch.autograd.grad( u, y, grad_outputs=torch.ones_like(u), create_graph=True, retain_graph=True, )[0] # 二阶空间导数 d2u_dx2 = torch.autograd.grad( du_dx, x, grad_outputs=torch.ones_like(du_dx), create_graph=True, retain_graph=True, )[0] d2u_dy2 = torch.autograd.grad( du_dy, y, grad_outputs=torch.ones_like(du_dy), create_graph=True, retain_graph=True, )[0] residual = du_dt - alpha * (d2u_dx2 + d2u_dy2) return torch.mean(residual ** 2)

这段代码是 PINN 的核心。它把神经网络变成一个“满足物理方程的代理模型”。每个训练批次会随机采样一组物理点,计算该点的残差损失,反向传播时同时更新 PINN 分支和融合层参数。

4.6 步骤五:训练主流程

# 文件路径:train.py import torch import torch.nn as nn import torch.optim as optim import numpy as np from data_gen import generate_thermal_field from dataset import build_samples, make_loader from model import PINNLSTMModel, LSTMOnlyModel from physics import physics_residual_loss def train_model(model, train_loader, alpha=0.02, lambda_phys=0.1, epochs=20, lr=1e-3, device="cpu"): model.to(device) optimizer = optim.Adam(model.parameters(), lr=lr) criterion = nn.MSELoss() for epoch in range(epochs): model.train() total_data_loss = 0.0 total_phys_loss = 0.0 for hist, t, x, y, target in train_loader: hist = hist.to(device) t = t.to(device) x = x.to(device) y = y.to(device) target = target.to(device) optimizer.zero_grad() if isinstance(model, PINNLSTMModel): pred, _, _ = model(hist, t, x, y) data_loss = criterion(pred, target) # 每个 batch 随机采样 512 个物理点 p_t = torch.rand(512, 1, device=device) * 1.0 p_x = torch.rand(512, 1, device=device) p_y = torch.rand(512, 1, device=device) phys_loss = physics_residual_loss(model, p_t, p_x, p_y, alpha) loss = data_loss + lambda_phys * phys_loss total_phys_loss += phys_loss.item() else: pred = model(hist) data_loss = criterion(pred, target) loss = data_loss loss.backward() optimizer.step() total_data_loss += data_loss.item() avg_data_loss = total_data_loss / len(train_loader) avg_phys_loss = total_phys_loss / len(train_loader) print(f"Epoch {epoch+1:02d} | Data Loss: {avg_data_loss:.6f} | Phys Loss: {avg_phys_loss:.6f}") return model if __name__ == "__main__": u = generate_thermal_field(nx=32, ny=32, nt=90, alpha=0.02) # 训练集:时间步 12~60 hist, t, x, y, label = build_samples(u[:, :, :], input_len=12) # 这里实际应改为按时间窗划分,下面会给出完整划分示例 train_loader = make_loader(hist, t, x, y, label, batch_size=256) # 训练 PINN+LSTM model = PINNLSTMModel() train_model(model, train_loader, epochs=20, device="cpu")

上面的build_samples全部取了一遍。更合理的做法是先按时间切分,再构造样本。下面给出完整的时间划分版本:

# 在 dataset.py 中增加按时间窗划分的功能 def build_samples_with_window(u, input_len=12, horizon=1, start_min=12, start_max=60): nt, nx, ny = u.shape xs = np.linspace(0, 1, nx) ys = np.linspace(0, 1, ny) hist_list, t_list, x_list, y_list, label_list = [], [], [], [], [] for i in range(nx): for j in range(ny): for start in range(start_min, min(start_max, nt - horizon)): hist = u[start - input_len:start, i, j] target = u[start + horizon, i, j] pred_time = (start + horizon) / nt hist_list.append(hist) t_list.append(pred_time) x_list.append(xs[i]) y_list.append(ys[j]) label_list.append(target) return ( np.array(hist_list, dtype=np.float32), np.array(t_list, dtype=np.float32).reshape(-1, 1), np.array(x_list, dtype=np.float32).reshape(-1, 1), np.array(y_list, dtype=np.float32).reshape(-1, 1), np.array(label_list, dtype=np.float32).reshape(-1, 1), ) # 训练集与测试集按时间窗切分 train_hist, train_t, train_x, train_y, train_label = \ build_samples_with_window(u, start_min=12, start_max=60) test_hist, test_t, test_x, test_y, test_label = \ build_samples_with_window(u, start_min=60, start_max=88)

4.7 步骤六:评估与对比

# 文件路径:evaluate.py import torch import numpy as np from sklearn.metrics import mean_squared_error, r2_score from dataset import build_samples_with_window, make_loader from model import PINNLSTMModel, LSTMOnlyModel def evaluate(model, loader, device="cpu"): model.eval() preds, trues = [], [] with torch.no_grad(): for hist, t, x, y, target in loader: hist = hist.to(device) t = t.to(device) x = x.to(device) y = y.to(device) if isinstance(model, PINNLSTMModel): pred, _, _ = model(hist, t, x, y) else: pred = model(hist) preds.append(pred.cpu().numpy()) trues.append(target.numpy()) preds = np.concatenate(preds).ravel() trues = np.concatenate(trues).ravel() rmse = mean_squared_error(trues, preds, squared=False) r2 = r2_score(trues, preds) return rmse, r2 if __name__ == "__main__": from data_gen import generate_thermal_field from dataset import make_loader import torch u = generate_thermal_field(nx=32, ny=32, nt=90, alpha=0.02) train_hist

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询