当前位置：首页 > news >正文

用PyTorch复现自适应动态规划HDP：一个非线性系统控制的保姆级代码解析

news 2026/5/10 20:36:45

用PyTorch实现自适应动态规划HDP：从理论到代码的深度解析

在控制理论与强化学习的交叉领域，自适应动态规划（Adaptive Dynamic Programming，ADP）正成为解决复杂非线性系统控制问题的有力工具。本文将聚焦HDP（Heuristic Dynamic Programming）这一经典ADP架构，通过PyTorch框架完整实现其核心算法，并深入剖析代码实现中的关键细节。不同于简单的代码展示，我们将从控制理论出发，逐步构建完整的神经网络实现方案，特别关注Actor-Critic架构在动态规划中的独特应用方式。

1. HDP理论基础与系统建模

HDP作为ADP家族中最基础的架构，其核心思想是通过三个神经网络的协同工作来逼近动态规划的最优解。这三个网络分别是：

模型网络（Model Network）：负责学习系统动力学方程
评价网络（Critic Network）：估计系统状态的价值函数
执行网络（Action Network）：生成最优控制策略

考虑一个典型的非线性控制系统示例：

def system_dynamics(x, u): dx1 = 0.2 * x[0] * np.exp(x[1]**2) dx2 = 0.3 * x[1]**3 - 0.2 * u return np.array([dx1, dx2])

这个系统表现出明显的非线性特性，传统线性控制方法难以处理。我们定义性能指标为：

$$ J = \frac{1}{2}\sum_{k=0}^{\infty} (x_k^T Q x_k + u_k^T R u_k) $$

其中Q和R通常取为单位矩阵，表示对状态和控制的同等重视。

2. 神经网络架构设计与实现

2.1 模型网络构建

模型网络需要学习系统状态转移函数，其输入为当前状态和控制输入，输出为下一时刻的状态预测。在PyTorch中实现如下：

class ModelNetwork(nn.Module): def __init__(self, state_dim=2, action_dim=1): super().__init__() self.fc1 = nn.Linear(state_dim + action_dim, 10) self.fc2 = nn.Linear(10, state_dim) def forward(self, x): x = F.relu(self.fc1(x)) return self.fc2(x)

关键训练技巧：

使用均方误差（MSE）作为损失函数
采用小批量梯度下降提高训练稳定性
训练数据应覆盖系统可能的状态空间

2.2 Critic和Actor网络设计

Critic网络评估状态价值，Actor网络生成控制策略，两者共同构成HDP的核心：

class CriticNetwork(nn.Module): def __init__(self, state_dim=2): super().__init__() self.fc1 = nn.Linear(state_dim, 10) self.fc2 = nn.Linear(10, 1) def forward(self, x): x = F.relu(self.fc1(x)) return self.fc2(x) class ActorNetwork(nn.Module): def __init__(self, state_dim=2, action_dim=1): super().__init__() self.fc1 = nn.Linear(state_dim, 10) self.fc2 = nn.Linear(10, action_dim) def forward(self, x): x = F.relu(self.fc1(x)) return torch.tanh(self.fc2(x)) * 10 # 限制输出范围

3. 训练流程与算法实现

HDP的训练分为三个阶段：模型网络预训练、Critic网络训练和Actor网络训练。下面是完整的训练循环实现：

def train_hdp(): # 初始化网络和优化器 model_net = ModelNetwork() critic_net = CriticNetwork() actor_net = ActorNetwork() # 第一阶段：模型网络训练 for epoch in range(model_train_epochs): # 生成训练数据 states = torch.rand(batch_size, 2) * torch.tensor([4, 2]) - torch.tensor([2, 1]) actions = torch.rand(batch_size, 1) * 20 - 10 inputs = torch.cat([states, actions], dim=1) targets = torch.stack([system_dynamics(s, a) for s, a in zip(states, actions)]) # 训练步骤 preds = model_net(inputs) loss = F.mse_loss(preds, targets) optimizer_model.zero_grad() loss.backward() optimizer_model.step() # 冻结模型网络参数 for param in model_net.parameters(): param.requires_grad = False # 第二阶段：Critic和Actor交替训练 for epoch in range(main_train_epochs): # Critic网络更新 states = torch.rand(batch_size, 2) * torch.tensor([4, 2]) - torch.tensor([2, 1]) actions = actor_net(states) next_states = model_net(torch.cat([states, actions], dim=1)) current_values = critic_net(states) next_values = critic_net(next_states.detach()) rewards = (states.pow(2).sum(1, keepdim=True) + actions.pow(2)) target_values = rewards + gamma * next_values critic_loss = F.mse_loss(current_values, target_values.detach()) optimizer_critic.zero_grad() critic_loss.backward() optimizer_critic.step() # Actor网络更新 actions = actor_net(states) next_states = model_net(torch.cat([states, actions], dim=1)) next_values = critic_net(next_states) actor_loss = -next_values.mean() # 最大化期望回报 optimizer_actor.zero_grad() actor_loss.backward() optimizer_actor.step()

4. 关键实现细节与调试技巧

在实际实现HDP算法时，有几个关键点需要特别注意：

模型网络的预训练质量：
- 确保训练数据覆盖系统所有工作区域
- 验证模型在边界状态的预测能力
- 使用标准化输入提高训练稳定性
Critic网络的训练技巧：
- 学习率设置要足够小以避免发散
- 采用目标网络稳定训练过程
- 定期验证Critic的价值估计是否合理
Actor网络的优化策略：
- 输出层使用tanh激活函数限制控制量范围
- 采用策略梯度方法更新参数
- 可以引入探索噪声提高策略多样性

常见问题排查表：

问题现象	可能原因	解决方案
Critic损失震荡	学习率过大	降低学习率或使用自适应优化器
控制策略收敛到局部最优	探索不足	增加训练初期的随机探索
模型预测误差大	训练数据不足	增加数据量或改进网络结构
训练过程不稳定	奖励尺度不合适	对奖励进行归一化处理

5. 实际应用与性能评估

完成训练后，我们可以评估HDP控制器在实际系统中的表现。下面是一个闭环控制的实现示例：

def simulate_control(actor, critic, model, initial_state, steps=100): state = torch.tensor(initial_state, dtype=torch.float32) states = [state.numpy()] actions = [] values = [] for _ in range(steps): with torch.no_grad(): action = actor(state.unsqueeze(0)).squeeze() value = critic(state.unsqueeze(0)).item() next_state = model(torch.cat([state, action.unsqueeze(0)])) actions.append(action.item()) values.append(value) states.append(next_state.numpy()) state = next_state return np.array(states), np.array(actions), np.array(values)

典型性能指标对比如下：