# 用Python实战Koopman算子：从非线性动力学到线性嵌入的深度学习实现（附完整代码） 如果你曾经被非线性系统的复杂性所困扰，看着那些难以预测的轨迹和混沌行为感到无从下手，那么Koopman算子理论可能会为你打开一扇新的大门。作为一名长期在工业界和学术界之间游走的算法工程师，我最初接触这个概念时，也被其“用线性工具解决非线性问题”的宏大愿景所吸引。然而，理论上的优美往往伴随着工程上的挑战——如何将抽象的算子理论转化为一行行可运行的代码，并真正用于预测一个摆动的钟摆、一段湍流的演化，甚至是一个复杂机械臂的运动轨迹？这正是本文想要解决的问题。 本文不是一篇理论综述，而是一份面向实践者的“工程手册”。我们将暂时放下繁复的数学推导，聚焦于如何用Python和深度学习框架，一步步构建一个能够学习并应用Koopman算子的模型。无论你是希望将先进控制理论应用于机器人项目的工程师，还是试图用数据驱动方法理解复杂生物系统的科研人员，只要具备基本的Python和机器学习知识，就能跟随本文的指引，亲手搭建起这座连接非线性世界与线性分析工具的桥梁。我们将从最基础的数据模拟开始，历经自编码器设计、线性动力学层集成、损失函数构建等关键环节，最终得到一个能够进行多步预测的完整模型，并提供所有可复现的代码。 ## 1. 核心理念：为什么我们需要Koopman算子？ 在深入代码之前，我们有必要用最直白的语言理解Koopman算子的核心价值。想象一下，你面前有一个单摆，它的运动方程是非线性的，其相空间轨迹是一个闭合的椭圆。传统的线性化方法（如在平衡点附近进行泰勒展开）只能在小角度摆动时给出近似，一旦摆幅增大，预测就会迅速失效。Koopman算子则提供了一种截然不同的思路：它不去直接线性化系统的状态空间，而是去寻找一组**观测函数**，将原始状态映射到一个新的空间。在这个新空间中，神奇的事情发生了——原本非线性的动力学演化，变成了一个简单的**线性变换**。 > 关键洞察：Koopman算子不改变物理世界的非线性本质，它改变的是我们“观察”世界的视角。通过找到正确的“眼镜”（观测函数），混沌的舞蹈可以看起来像匀速的旋转。 这种方法的优势是显而易见的： * **解锁强大的线性工具库**：一旦系统在提升后的空间中是线性的，我们就可以毫无顾忌地使用特征值分解、模态分析、线性最优控制（LQR）等一系列成熟、高效的理论工具。 * **全局性潜力**：与局部线性化不同，一个理想的Koopman嵌入旨在实现**全局线性化**，即在整个状态空间范围内都保持线性动力学关系。 * **数据驱动友好**：我们不需要知道系统精确的微分方程。通过观测系统产生的时间序列数据，我们可以用机器学习的方法直接学习出这组观测函数和线性动力学矩阵。 然而，最大的挑战也在于此：如何找到这组“神奇”的观测函数？传统方法依赖于专家知识手工设计基函数（如多项式、径向基函数），这对于简单系统尚可，面对高维、强非线性系统则力不从心。这正是深度学习大显身手的地方。 ## 2. 工程蓝图：深度Koopman模型的整体架构 我们的目标是用神经网络自动学习Koopman嵌入。整个模型的架构可以看作一个“编码-演化-解码”的过程，其核心组件如下图所示（此处为概念描述）： 1. **编码器 (Encoder)**：一个神经网络，充当我们要寻找的观测函数。它接收原始系统状态 `x(t)`（例如，单摆的角度和角速度），并将其映射到所谓的 **Koopman 空间** 中的坐标 `y(t)`。我们希望这个空间维度尽可能低，且动力学是线性的。 2. **线性动力学层 (Koopman Matrix)**：这是一个关键的线性层，通常表示为一个矩阵 `K`。它负责在 Koopman 空间中推进时间：`y(t+Δt) = K * y(t)`。这个矩阵就是我们对无限维 Koopman 算子的有限维近似。 3. **解码器 (Decoder)**：另一个神经网络，是编码器的近似逆。它将 Koopman 空间中的坐标 `y(t)` 映射回原始状态空间，重建出 `x_recon(t)`。解码器的存在确保了我们的嵌入没有丢失重建原始状态所需的信息。 整个模型通过一个自编码器结构进行端到端训练，但损失函数远不止重建误差。为了强制 Koopman 空间中的线性动力学，我们需要引入基于时间序列的预测损失。 下面，我们用代码来具体定义这个架构。我们将使用 PyTorch 作为深度学习框架。 ```python import torch import torch.nn as nn import numpy as np class DeepKoopman(nn.Module): """ 深度 Koopman 自编码器模型。 假设输入状态维度为 n， Koopman 空间维度为 k。 """ def __init__(self, n=2, k=10, hidden_dims=[64, 32]): super(DeepKoopman, self).__init__() self.n = n # 原始状态维度 self.k = k # Koopman 嵌入维度 # --- 编码器网络 --- encoder_layers = [] prev_dim = n for h_dim in hidden_dims: encoder_layers.append(nn.Linear(prev_dim, h_dim)) encoder_layers.append(nn.ReLU()) prev_dim = h_dim encoder_layers.append(nn.Linear(prev_dim, k)) self.encoder = nn.Sequential(*encoder_layers) # --- Koopman 矩阵 K (线性动力学层) --- # 我们将其初始化为一个近似的单位矩阵，表示初始假设动力学变化很慢 self.K = nn.Parameter(torch.eye(k) * 0.95 + torch.randn(k, k) * 0.05) # --- 解码器网络 --- decoder_layers = [] prev_dim = k for h_dim in reversed(hidden_dims): decoder_layers.append(nn.Linear(prev_dim, h_dim)) decoder_layers.append(nn.ReLU()) prev_dim = h_dim decoder_layers.append(nn.Linear(prev_dim, n)) self.decoder = nn.Sequential(*decoder_layers) def encode(self, x): """将原始状态 x 映射到 Koopman 空间 y""" return self.encoder(x) def decode(self, y): """将 Koopman 空间坐标 y 重建为原始状态 x""" return self.decoder(y) def koopman_evolve(self, y, steps=1): """ 在 Koopman 空间中进行线性演化。 y: 初始 Koopman 坐标 steps: 向前推进的步数 返回: 演化后的 Koopman 坐标序列 [y, K*y, K^2*y, ...] """ y_seq = [y] for _ in range(steps): y = torch.matmul(y, self.K.T) # 注意维度匹配，这里假设 y 是 (batch, k) y_seq.append(y) return torch.stack(y_seq, dim=1) # 形状: (batch, steps+1, k) def forward(self, x, prediction_steps=0): """ 前向传播。 如果 prediction_steps=0，则只进行编码-解码（重建）。 如果 prediction_steps>0，则额外进行多步预测。 """ y0 = self.encode(x) # 初始嵌入 x_recon = self.decode(y0) # 重建 if prediction_steps > 0: # 进行多步演化 y_evolution = self.koopman_evolve(y0, steps=prediction_steps) # (batch, steps+1, k) # 将每一步的 Koopman 坐标解码为状态 batch_size, seq_len, k_dim = y_evolution.shape x_pred = self.decode(y_evolution.reshape(-1, k_dim)).reshape(batch_size, seq_len, self.n) return x_recon, x_pred else: return x_recon ``` 这个 `DeepKoopman` 类定义了我们模型的基本骨架。`encode` 和 `decode` 函数构成了自编码器，而 `koopman_evolve` 函数利用可学习的矩阵 `K` 实现了线性动力学演化。`forward` 函数根据参数决定是进行单步重建还是多步预测。 ## 3. 数据准备：模拟非线性动力学系统 为了训练模型，我们需要数据。我们将从几个经典的、有解析解的非线性系统中生成模拟数据，这样便于我们验证模型的学习效果。这里我们实现一个简单的数据生成器。 ```python def generate_duffing_oscillator(num_trajectories=100, length=50, dt=0.1): """ 生成 Duffing 振子的时间序列数据。 方程: dx/dt = y, dy/dt = x - x^3 - δ*y + γ*cos(ω*t) 这是一个具有双势阱的非线性阻尼受迫振子。 """ delta, gamma, omega = 0.1, 0.37, 1.0 data = [] for _ in range(num_trajectories): # 随机初始条件 x0 = np.random.uniform(-2, 2, 2) traj = [x0] for i in range(1, length): x, y = traj[-1] dx = y dy = x - x**3 - delta * y + gamma * np.cos(omega * i * dt) x_new = x + dx * dt y_new = y + dy * dt traj.append([x_new, y_new]) data.append(traj) return np.array(data) # 形状: (num_trajectories, length, 2) def generate_lorenz_system(num_trajectories=50, length=100, dt=0.02): """ 生成 Lorenz 系统的时间序列数据。 方程: dx/dt = σ*(y - x), dy/dt = x*(ρ - z) - y, dz/dt = x*y - β*z 经典参数 (σ=10, ρ=28, β=8/3) 会产生混沌行为。 """ sigma, rho, beta = 10.0, 28.0, 8.0/3.0 data = [] for _ in range(num_trajectories): # 在吸引子附近随机初始化 x0 = np.random.uniform(-20, 20, 3) traj = [x0] for i in range(1, length): x, y, z = traj[-1] dx = sigma * (y - x) dy = x * (rho - z) - y dz = x * y - beta * z x_new = x + dx * dt y_new = y + dy * dt z_new = z + dz * dt traj.append([x_new, y_new, z_new]) data.append(traj) return np.array(data) # 形状: (num_trajectories, length, 3) # 数据预处理：创建用于训练的输入-目标对 def create_sequence_dataset(data, window_size): """ 将长序列切割成用于训练的重叠窗口。 data: 形状 (num_trajectories, seq_len, state_dim) window_size: 每个训练样本的序列长度 (T) 返回: inputs (..., T-1, state_dim), targets (..., T-1, state_dim) 其中 targets 是 inputs 向后一步的序列。 """ inputs, targets = [], [] for traj in data: for i in range(len(traj) - window_size): window = traj[i:i+window_size] inputs.append(window[:-1]) # 从 t0 到 t_{T-2} targets.append(window[1:]) # 从 t1 到 t_{T-1} return np.array(inputs), np.array(targets) ``` 我们选择了Duffing振子和Lorenz系统作为例子，因为它们分别代表了具有复杂周期行为和混沌行为的经典非线性系统。`create_sequence_dataset` 函数将长序列切割成连续的短窗口，这是训练时序预测模型的常见做法。 ## 4. 损失函数设计：模型训练的灵魂 如何训练我们的 DeepKoopman 模型？损失函数的设计至关重要，它需要同时满足多个目标： 1. **重建精度**：编码后再解码的状态应尽可能接近原始状态。 2. **线性动力学一致性**：在 Koopman 空间中，一步和多步的线性演化应能准确预测未来状态。 3. **模型简洁性**：避免过拟合，鼓励学习到本质的、低维的动力学。 我们定义一个多任务损失函数： ```python def koopman_loss(model, x_sequence, lambda_rec=1.0, lambda_pred=1.0, lambda_reg=1e-4): """ 计算 DeepKoopman 模型的损失。 x_sequence: 一个批量的输入序列，形状 (batch_size, seq_len, state_dim) seq_len = T，我们使用前 T-1 步作为输入，预测后 T-1 步。 """ batch_size, seq_len, state_dim = x_sequence.shape device = next(model.parameters()).device x_sequence = x_sequence.to(device) # 输入是序列的前 T-1 步 x_input = x_sequence[:, :-1, :] # (batch, T-1, n) # 目标是序列的后 T-1 步（即下一步的状态） x_target = x_sequence[:, 1:, :] # (batch, T-1, n) # 将输入序列展平，用于编码 x_input_flat = x_input.reshape(-1, state_dim) # (batch*(T-1), n) y0_flat = model.encode(x_input_flat) # (batch*(T-1), k) # 1. 重建损失 x_recon_flat = model.decode(y0_flat) reconstruction_loss = nn.functional.mse_loss(x_recon_flat, x_input_flat) # 2. 单步预测损失 (在 Koopman 空间中) # 对每个时间点，用 K 矩阵推进一步 y1_pred_flat = torch.matmul(y0_flat, model.K.T) x1_pred_flat = model.decode(y1_pred_flat) # 注意：x1_pred_flat 应该与 x_target 的展平形式比较 x_target_flat = x_target.reshape(-1, state_dim) prediction_loss = nn.functional.mse_loss(x1_pred_flat, x_target_flat) # 3. 多步预测损失 (可选，更强制线性动力学) # 我们可以取每个序列的第一个点，用模型预测整个后续序列，然后与真实序列比较 x_first = x_sequence[:, 0, :] # (batch, n) y_first = model.encode(x_first) # (batch, k) # 演化 T-1 步 y_evol = model.koopman_evolve(y_first, steps=seq_len-1) # (batch, seq_len, k) batch_sz, seq_len_k, k_dim = y_evol.shape x_pred_full = model.decode(y_evol.reshape(-1, k_dim)).reshape(batch_sz, seq_len_k, state_dim) multi_step_loss = nn.functional.mse_loss(x_pred_full, x_sequence[:, :seq_len_k, :]) # 4. 正则化损失 (防止 K 矩阵变得病态) # Frobenius 范数正则化 regularization_loss = torch.norm(model.K, p='fro') # 总损失 total_loss = (lambda_rec * reconstruction_loss + lambda_pred * prediction_loss + 0.1 * lambda_pred * multi_step_loss + # 多步损失权重可以小一些 lambda_reg * regularization_loss) return total_loss, reconstruction_loss, prediction_loss, multi_step_loss ``` 这个损失函数是模型成功的关键。`reconstruction_loss` 确保自编码器有效；`prediction_loss` 强制单步线性动力学；`multi_step_loss` 进一步约束多步预测的准确性，这是 Koopman 线性假设的严格检验；`regularization_loss` 则帮助稳定训练。 ## 5. 训练流程与实战技巧 有了模型、数据和损失函数，我们可以开始训练了。这里给出一个完整的训练循环，并分享一些在实际项目中积累的技巧。 ```python import torch.optim as optim from torch.utils.data import DataLoader, TensorDataset # 1. 生成并准备数据 print("生成训练数据...") train_data = generate_duffing_oscillator(num_trajectories=200, length=30) # train_data = generate_lorenz_system(num_trajectories=100, length=50) # 可以尝试 Lorenz 系统 inputs_np, targets_np = create_sequence_dataset(train_data, window_size=10) # 转换为 PyTorch 张量 inputs = torch.FloatTensor(inputs_np) targets = torch.FloatTensor(targets_np) train_dataset = TensorDataset(inputs, targets) train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True) # 2. 初始化模型、优化器 state_dim = inputs_np.shape[-1] koopman_dim = 8 # 尝试不同的嵌入维度 model = DeepKoopman(n=state_dim, k=koopman_dim, hidden_dims=[32, 16]) optimizer = optim.Adam(model.parameters(), lr=1e-3) scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=100, gamma=0.9) # 3. 训练循环 num_epochs = 300 device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') model.to(device) print(f"开始训练，设备: {device}") for epoch in range(num_epochs): model.train() total_loss_accum = 0.0 for batch_inputs, batch_targets in train_loader: # 注意：我们的损失函数需要完整的序列，而 DataLoader 给出的是 (batch, T-1, n) # 我们需要用 inputs 和 targets 拼接出完整的序列窗口 batch_inputs = batch_inputs.to(device) batch_targets = batch_targets.to(device) # 构造完整序列：取 inputs 的第一帧，加上 targets 的最后一帧（或直接使用 inputs 和 targets 的对应关系） # 这里我们采用一种简单方式：将 batch_inputs 作为序列的前 T-1 步，用其最后一帧的预测作为第 T 步（近似） # 更严谨的做法是在数据生成时保留完整窗口。 batch_seq = torch.cat([batch_inputs, batch_targets[:, -1:, :]], dim=1) # (batch, T, n) optimizer.zero_grad() total_loss, rec_loss, pred_loss, multi_loss = koopman_loss(model, batch_seq, lambda_rec=1.0, lambda_pred=1.0, lambda_reg=1e-5) total_loss.backward() # 梯度裁剪，防止训练不稳定 torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) optimizer.step() total_loss_accum += total_loss.item() scheduler.step() avg_loss = total_loss_accum / len(train_loader) if (epoch + 1) % 50 == 0: print(f"Epoch [{epoch+1:04d}/{num_epochs}] | Avg Loss: {avg_loss:.6f} | LR: {scheduler.get_last_lr()[0]:.6f}") # 可以在这里添加验证集评估和模型保存逻辑 print("训练完成！") ``` **实战技巧与常见陷阱：** * **嵌入维度 `k` 的选择**：这是一个超参数。太小会导致信息丢失，预测不准；太大会增加学习难度，且 `K` 矩阵可能学到虚假的、非线性的模式（在子空间内）。通常从与状态维度 `n` 相当或稍大的值开始尝试。 * **Koopman 矩阵 `K` 的初始化**：初始化为接近单位矩阵是个好策略，这假设初始动力学变化缓慢。随机初始化可能导致训练初期不稳定。 * **损失权重平衡**：`lambda_rec`, `lambda_pred`, `lambda_reg` 需要调优。如果重建损失太大，模型可能变成一个普通的自编码器，忽略动力学；如果预测损失太大，重建质量可能下降，影响解码精度。我的经验是从 `1:1` 开始，根据验证集上的预测性能和重建误差进行调整。 * **多步预测的重要性**：在损失函数中加入 `multi_step_loss` 对于学习到**真正线性**的动力学至关重要。只优化单步预测，模型可能会“作弊”，例如学习一个恒等映射的编码，然后靠解码器非线性地拟合下一步，这完全违背了 Koopman 的初衷。 * **梯度爆炸与消失**：由于我们通过 `K` 矩阵进行幂次乘法（`K^t`），如果 `K` 的特征值不在单位圆附近，多步预测时梯度可能会爆炸或消失。梯度裁剪和合适的正则化有助于缓解。 ## 6. 模型评估与可视化：看看它学到了什么 训练完成后，我们需要评估模型是否真的学到了有意义的 Koopman 嵌入和线性动力学。 ```python import matplotlib.pyplot as plt def evaluate_and_visualize(model, test_trajectory, device='cpu'): """ 在一条测试轨迹上评估模型，并进行可视化。 test_trajectory: 形状 (full_length, state_dim) """ model.eval() with torch.no_grad(): traj_tensor = torch.FloatTensor(test_trajectory).unsqueeze(0).to(device) # (1, L, n) full_length = traj_tensor.shape[1] # 选择一段进行预测 start_idx = 0 observe_len = 10 predict_len = 30 observed = traj_tensor[:, start_idx:start_idx+observe_len, :] # (1, observe_len, n) true_future = traj_tensor[:, start_idx+observe_len:start_idx+observe_len+predict_len, :] # (1, predict_len, n) # 用观察到的最后一点作为初始状态，进行多步预测 last_observed = observed[:, -1:, :] # (1, 1, n) y0 = model.encode(last_observed.squeeze(1)) # (1, k) y_pred_seq = model.koopman_evolve(y0, steps=predict_len-1) # (1, predict_len, k) x_pred_seq = model.decode(y_pred_seq.reshape(-1, model.k)).reshape(1, predict_len, model.n) # 转换为 numpy 用于绘图 observed_np = observed.squeeze(0).cpu().numpy() true_future_np = true_future.squeeze(0).cpu().numpy() pred_future_np = x_pred_seq.squeeze(0).cpu().numpy() # 绘图 fig, axes = plt.subplots(1, 2, figsize=(14, 5)) # 子图1：状态空间轨迹对比 ax1 = axes[0] ax1.plot(observed_np[:, 0], observed_np[:, 1], 'bo-', label='Observed', linewidth=2, markersize=6) ax1.plot(true_future_np[:, 0], true_future_np[:, 1], 'g--', label='True Future', linewidth=2) ax1.plot(pred_future_np[:, 0], pred_future_np[:, 1], 'r-.', label='Predicted Future', linewidth=2) ax1.scatter(observed_np[-1, 0], observed_np[-1, 1], c='black', s=100, zorder=5, label='Start Prediction') ax1.set_xlabel('State x1') ax1.set_ylabel('State x2') ax1.set_title('Phase Space Trajectory') ax1.legend() ax1.grid(True, alpha=0.3) ax1.axis('equal') # 子图2：时间序列对比 ax2 = axes[1] time_obs = np.arange(observe_len) time_future = np.arange(observe_len, observe_len + predict_len) ax2.plot(time_obs, observed_np[:, 0], 'bo-', label='Observed x1', linewidth=2) ax2.plot(time_future, true_future_np[:, 0], 'g--', label='True Future x1', linewidth=2) ax2.plot(time_future, pred_future_np[:, 0], 'r-.', label='Predicted Future x1', linewidth=2) ax2.set_xlabel('Time Step') ax2.set_ylabel('State Value') ax2.set_title('Time Series Prediction') ax2.legend() ax2.grid(True, alpha=0.3) plt.tight_layout() plt.show() # 计算预测误差 mse = np.mean((true_future_np - pred_future_np) ** 2) print(f"多步预测 ({predict_len} 步) MSE: {mse:.6f}") # 分析 Koopman 矩阵 K K_np = model.K.detach().cpu().numpy() print(f"\nKoopman 矩阵 K (部分):") print(K_np[:5, :5]) # 打印前5x5 eigvals, eigvecs = np.linalg.eig(K_np) print(f"\nK 矩阵的特征值 (模): {np.abs(eigvals)}") # 特征值的模接近1，表示动力学在 Koopman 空间中是稳定的（无发散/过快衰减）。 # 生成一条测试轨迹并评估 test_traj = generate_duffing_oscillator(num_trajectories=1, length=100)[0] evaluate_and_visualize(model, test_traj, device=device) ``` 可视化是理解模型行为的利器。相位空间轨迹图能直观展示预测轨迹是否遵循真实的动力学流形。时间序列图则清晰显示了预测误差如何随时间累积。此外，分析 Koopman 矩阵 `K` 的特征值也很有意义：在连续时间系统中，其特征值应位于复平面单位圆上或附近，对应着振荡模式；衰减或增长的模式则对应着特征值模小于或大于1。 ## 7. 进阶话题与扩展方向 一个基础的 DeepKoopman 模型已经搭建完成，但要将其应用于更复杂、更实际的场景，还需要考虑以下进阶问题： **处理连续谱与外部强迫项** 许多实际系统（如湍流）具有连续的特征值谱，或者受随时间变化的外部输入驱动。基础的线性矩阵 `K` 难以描述这些。一种扩展是引入**参数化的 Koopman 算子**，例如，让 `K` 成为系统状态或外部输入的函数（通过一个小型神经网络生成），或者引入一个额外的“控制输入”通道。 ```python class ParametricKoopmanLayer(nn.Module): """一个简单的参数化 Koopman 层示例，K 矩阵由当前状态的一个函数生成。""" def __init__(self, state_dim, koopman_dim, hidden_dim=32): super().__init__() self.k = koopman_dim self.net = nn.Sequential( nn.Linear(state_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, koopman_dim * koopman_dim) ) def forward(self, x_state): # x_state: (batch, state_dim) batch_size = x_state.shape[0] K_flat = self.net(x_state) # (batch, k*k) K = K_flat.view(batch_size, self.k, self.k) # (batch, k, k) return K # 返回一个批量的 K 矩阵 ``` **提升可解释性：关联物理量** 我们学习到的 Koopman 坐标 `y` 通常是抽象的。为了提升可解释性，可以在损失函数中加入约束，使某些特定的 `y` 分量与我们有物理意义的量（如能量、动量）相关联。这属于“物理信息神经网络”的范畴。 **从仿真到真实世界数据的挑战** 在仿真中，数据无噪声、采样均匀。真实世界数据则充满挑战： * **噪声**：需要在损失函数中加入对噪声鲁棒的设计，或使用去噪自编码器作为编码器。 * **非均匀采样**：Koopman 矩阵 `K` 对应固定的时间步长 `Δt`。如果数据采样不均匀，需要将 `Δt` 作为参数融入模型，或者使用连续时间框架（学习一个无穷小生成元 `L`，使得 `K ≈ exp(L*Δt)`）。 * **高维观测**：当输入是图像或视频时（如流体流动的 PIV 图像），编码器需要换成 CNN 或 Vision Transformer，解码器则需要对应地上采样。核心的线性动力学层保持不变，这体现了 Koopman 方法的灵活性。 **部署与实时控制** 最终，我们可能希望将学习到的 Koopman 模型用于实时预测或控制。这要求： 1. **模型轻量化**：简化编码器/解码器网络，或使用知识蒸馏将大模型压缩为小模型。 2. **与控制器集成**：利用 Koopman 空间的线性特性，设计线性二次型调节器。模型需要能够快速计算状态 `x` 对应的 Koopman 坐标 `y`，然后根据线性系统理论计算控制量 `u`，再通过一个额外的网络或映射将 `u` 的影响反映到状态演化中。 我在一个机械臂关节角度预测的小项目中尝试过最后一点。将训练好的 Koopman 模型封装成一个 ROS 节点，它实时接收关节角度和速度，预测未来几秒内的轨迹，并发送给一个基于线性模型的预测控制器。最大的收获是，**一定要在损失函数里加强对多步预测的约束**，否则学到的“线性”动力学在控制回路中很快就会因误差累积而失效。