神经算子在计算流体动力学中的应用

1. 引言

计算流体动力学（Computational Fluid Dynamics, CFD）是科学计算的核心应用领域之一。传统的CFD方法基于有限差分、有限元或有限体积法，计算成本高昂。神经算子提供了一种数据驱动的方法来加速CFD模拟，同时保持较高的预测精度¹。

---

2. CFD问题的数学框架

2.1 Navier-Stokes方程

不可压缩Navier-Stokes方程：

$$\frac{\partial \mathbf{u}}{\partial t}+(\mathbf{u}\cdot \nabla )\mathbf{u}=-\nabla p+\nu \Delta \mathbf{u}+\mathbf{f}$$ $$\nabla \cdot \mathbf{u}=0$$

其中：

• $\mathbf{u}$：速度场
• $p$：压力场
• $\nu$：运动粘度
• $\mathbf{f}$：外力

2.2 神经算子视角

将CFD问题形式化为算子学习：

$$\mathcal{G}:{\mathbf{u}}_0(\mathbf{x})\to \mathbf{u}(\mathbf{x},t)$$

或

$$\mathcal{G}:\mathbf{f}(\mathbf{x})\to \mathbf{u}(\mathbf{x})$$

---

3. 主流应用场景

3.1 流场预测

任务：从初始/边界条件预测完整流场

class FlowFieldPredictor(nn.Module):
    """
    基于FNO的流场预测器
    """
    def __init__(self, modes, width, n_vars=3):
        super().__init__()
        self.n_vars = n_vars  # u, v, p
        
        self.fno = FNO2d(modes, modes, width * n_vars)
        
        # 输出解码器
        self.decoder = nn.Sequential(
            nn.Linear(width * n_vars, width),
            nn.GELU(),
            nn.Linear(width, n_vars)
        )
    
    def forward(self, x):
        # x: (batch, H, W, n_vars)
        features = self.fno(x)
        return self.decoder(features)

3.2 湍流建模

任务：学习雷诺应力项或亚网格尺度模型

class TurbulenceModel(nn.Module):
    """
    神经亚网格应力模型
    """
    def __init__(self, input_channels, hidden_dim):
        super().__init__()
        # 条件输入：局部平均速度
        self.condition_encoder = nn.Sequential(
            nn.Conv2d(2, hidden_dim, 3, padding=1),
            nn.GELU()
        )
        
        # 主干网络
        self.backbone = nn.Sequential(
            nn.Conv2d(input_channels + hidden_dim, hidden_dim, 3, padding=1),
            nn.GELU(),
            nn.Conv2d(hidden_dim, hidden_dim, 3, padding=1),
            nn.GELU(),
            nn.Conv2d(hidden_dim, 1, 3, padding=1)  # 输出：亚网格应力
        )
    
    def forward(self, u_fine, u_coarse):
        """
        u_fine: 细网格速度
        u_coarse: 粗网格速度
        """
        # 局部平均场
        local_mean = F.avg_pool2d(u_fine, 4, stride=4)
        local_mean = F.interpolate(local_mean, scale_factor=4, mode='bilinear')
        
        # 条件编码
        cond = self.condition_encoder(
            torch.cat([u_coarse, local_mean], dim=1)
        )
        
        # 预测亚网格应力
        tau_sgs = self.backbone(
            torch.cat([u_fine, cond], dim=1)
        )
        
        return tau_sgs

3.3 流体参数推断

任务：从观测数据推断流体参数（如粘度、阻力系数）

class ParameterInference(nn.Module):
    """
    基于DeepONet的参数推断器
    """
    def __init__(self, observation_channels, param_dim, hidden_dim):
        super().__init__()
        
        # 观测编码器
        self.obs_encoder = DeepONetBranch([64, 64, 64], hidden_dim)
        
        # 参数解码器
        self.param_decoder = nn.Sequential(
            nn.Linear(hidden_dim, hidden_dim),
            nn.GELU(),
            nn.Linear(hidden_dim, param_dim),
            nn.Sigmoid()  # 参数范围 [0, 1]
        )
    
    def forward(self, observations, positions):
        # observations: (batch, n_obs) - 观测点的速度值
        # positions: (batch, n_obs, 2) - 观测点位置
        
        features = self.obs_encoder(observations)
        params = self.param_decoder(features)
        
        return params

---

4. 典型架构配置

4.1 FNO配置

参数	推荐值	说明
modes	32-64	频谱模式数
width	32-64	通道维度
n_layers	4-8	Fourier层数
activation	GELU	激活函数

fno_config = {
    'modes1': 48,
    'modes2': 48,
    'width': 64,
    'n_layers': 6,
    'activation': 'gelu'
}

4.2 DeepONet配置

deeponet_config = {
    'branch_layers': [128, 128, 128],
    'trunk_layers': [128, 128, 128],
    'p': 256,  # 特征维度
    'activation': 'tanh'
}

4.3 训练配置

training_config = {
    'batch_size': 16,
    'epochs': 500,
    'learning_rate': 1e-3,
    'weight_decay': 1e-4,
    'scheduler': 'cosine',
    'early_stopping_patience': 50
}

---

5. 基准数据集

5.1 2D Navier-Stokes

• 数据集：2D Kolmogorov Flow, Cylinder Flow
• 分辨率：$256\times 256$ 到 $512\times 512$
• 任务：短期/长期预测

5.2 3D Turbulence

• 数据集： decaying turbulence, forced turbulence
• 分辨率：$64^3$ 到 $256^3$
• 任务：能谱预测

5.3 Airfoil Flow

• 数据集：NACA airfoil simulations
• 任务：升力/阻力预测

---

6. 性能评估

6.1 评估指标

指标	定义	用途
RelL2	$\frac{\Vert u_{pred}-u_{true}{\Vert }_2}{\Vert u_{true}{\Vert }_2}$	整体精度
MaxAE	$\max	u_{pred} - u_{true}
Energy spectrum	$	\hat{u}(k)
Divergence error	$\Vert \nabla \cdot u\Vert$	不可压性

6.2 典型性能

方法	RelL2误差	加速比
传统CFD	0 (基准)	1x
FNO	0.01-0.05	100-1000x
DeepONet	0.02-0.08	50-500x
PINO	0.005-0.02	50-500x

---

7. 物理约束集成

7.1 不可压缩性约束

class IncompressibilityLoss(nn.Module):
    def __init__(self, method='divergence'):
        super().__init__()
        self.method = method
    
    def forward(self, u_pred, x):
        if self.method == 'divergence':
            # 计算散度
            du_dx = torch.autograd.grad(u_pred[..., 0], x[..., 0], 
                                        grad_outputs=torch.ones_like(u_pred[..., 0]),
                                        create_graph=True)[0]
            dv_dy = torch.autograd.grad(u_pred[..., 1], x[..., 1],
                                        grad_outputs=torch.ones_like(u_pred[..., 1]),
                                        create_graph=True)[0]
            div = du_dx + dv_dy
            return torch.mean(div**2)
        
        elif self.method == 'penalty':
            # 压力-速度耦合
            pass

7.2 能量守恒约束

class EnergyConservationLoss(nn.Module):
    def forward(self, u_pred, u0):
        # 动能变化
        E_t = 0.5 * torch.mean(u_pred**2)
        E_0 = 0.5 * torch.mean(u0**2)
        
        # 对于保守系统
        return (E_t - E_0)**2

7.3 边界条件

class BoundaryConditionLoss(nn.Module):
    def __init__(self, bc_mask):
        super().__init__()
        self.register_buffer('bc_mask', bc_mask)
    
    def forward(self, u_pred, u_bc):
        return torch.mean(
            (u_pred[self.bc_mask] - u_bc)**2
        )

---

8. 实践案例

8.1 圆柱绕流

class CylinderFlowPredictor(nn.Module):
    """
    圆柱绕流的神经算子预测器
    """
    def __init__(self):
        super().__init__()
        # 编码器：处理流场初始状态
        self.encoder = nn.Sequential(
            nn.Conv2d(3, 32, 3, padding=1),  # u, v, p
            nn.GELU(),
            nn.Conv2d(32, 64, 3, padding=1),
            nn.GELU()
        )
        
        # FNO核心
        self.fno = FNO2d(modes1=32, modes2=32, width=64, n_layers=6)
        
        # 解码器
        self.decoder = nn.Sequential(
            nn.Conv2d(64, 32, 3, padding=1),
            nn.GELU(),
            nn.Conv2d(32, 3, 3, padding=1)
        )
    
    def forward(self, x, t_pred):
        """
        x: 初始流场
        t_pred: 预测时间
        """
        # 编码
        features = self.encoder(x)
        
        # 时间编码
        t_enc = self.time_encoder(t_pred)
        features = features + t_enc
        
        # FNO处理
        output = self.fno(features)
        
        # 解码
        return self.decoder(output)

8.2 多尺度湍流

class MultiScaleTurbulenceModel(nn.Module):
    """
    多尺度湍流模型
    结合WNO和FNO处理不同尺度的湍流特征
    """
    def __init__(self):
        super().__init__()
        
        # WNO处理大尺度结构
        self.wno_large = WaveletNeuralOperator(
            scales=[1, 2, 4, 8],
            width=64
        )
        
        # FNO处理小尺度湍流
        self.fno_small = FNO2d(
            modes1=64, modes2=64, width=32
        )
        
        # 尺度交互
        self.scale_interaction = CrossScaleAttention(
            dim=64, num_scales=4
        )
    
    def forward(self, u):
        # 大尺度
        u_large = self.wno_large(u)
        
        # 小尺度残余
        u_residual = u - u_large
        u_small = self.fno_small(u_residual)
        
        # 尺度融合
        return u_large + u_small

---

9. 挑战与展望

9.1 当前挑战

挑战	描述	可能的解决方案
长期预测稳定性	误差累积导致物理不一致	课程学习、重正化
高雷诺数湍流	湍流细节丢失	多尺度建模、超分辨率
非结构化网格	实际工程问题	GNO、MeshGraphNet
不确定性量化	可靠的不确定性估计	贝叶斯方法、集成

9.2 未来方向

1. Foundation Models for CFD：预训练CFD模型
2. 实时预测：嵌入式部署
3. 数字孪生：与物理系统闭环控制
4. 可解释性：物理一致性保证

---

10. 参考文献

---

相关主题

• 神经算子理论基础
• Fourier神经算子详解
• DeepONet深度解析
• AI在气候科学中的应用
• 科学机器学习索引

Footnotes

1. Kochkov, D., et al. (2024). Machine learning–accelerated computational fluid dynamics. PNAS. ↩