Python(PyTorch)物理变化可微分神经算法

本文主要是介绍Python(PyTorch)物理变化可微分神经算法，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

🎯要点

🍪语言内容分比

在这里插入图片描述

🍇PyTorch可微分优化

假设张量 $x$ 是元参数， $a$ 是普通参数（例如网络参数）。我们有内部损失 $^{\text {in }}=a_0 \cdot x^2$ 并且我们使用梯度 $\frac{\partial L ^{\text {in }}}{\partial a_0}=x^2$ 更新 $a$ 和 $a_1=a_0-\eta \frac{\partial L ^{\text {in }}}{\partial a_0}=a_0-\eta x^2$ 。然后我们计算外部损失 $^{\text {out }}=a_1 \cdot x^2$ 。因此外部损失到 $x$ 的梯度为：
$\begin{aligned} \frac{\partial L ^{\text {out }}}{\partial x} & =\frac{\partial\left(a_1 \cdot x^2\right)}{\partial x} \\ & =\frac{\partial a_1}{\partial x} \cdot x^2+a_1 \cdot \frac{\partial\left(x^2\right)}{\partial x} \\ & =\frac{\partial\left(a_0-\eta x^2\right)}{\partial x} \cdot x^2+\left(a_0-\eta x^2\right) \cdot 2 x \\ & =(-\eta \cdot 2 x) \cdot x^2+\left(a_0-\eta x^2\right) \cdot 2 x \\ & =-4 \eta x^3+2 a_0 x \end{aligned}$
鉴于上述分析解，让我们使用 TorchOpt 中的 MetaOptimizer 对其进行验证。MetaOptimizer 是我们可微分优化器的主类。它与功能优化器 torchopt.sgd 和 torchopt.adam 相结合，定义了我们的高级 API torchopt.MetaSGD 和 torchopt.MetaAdam。

首先，定义网络。

from IPython.display import displayimport torch
import torch.nn as nn
import torch.nn.functional as Fimport torchoptclass Net(nn.Module):def __init__(self):super().__init__()self.a = nn.Parameter(torch.tensor(1.0), requires_grad=True)def forward(self, x):return self.a * (x**2)

然后我们声明网络（由 a 参数化）和元参数 x。不要忘记为 x 设置标志 require_grad=True 。

net = Net()
x = nn.Parameter(torch.tensor(2.0), requires_grad=True)

接下来我们声明元优化器。这里我们展示了定义元优化器的两种等效方法。

optim = torchopt.MetaOptimizer(net, torchopt.sgd(lr=1.0))
optim = torchopt.MetaSGD(net, lr=1.0)

元优化器将网络作为输入并使用方法步骤来更新网络（由a参数化）。最后，我们展示双层流程的工作原理。

inner_loss = net(x)
optim.step(inner_loss)outer_loss = net(x)
outer_loss.backward()
# x.grad = - 4 * lr * x^3 + 2 * a_0 * x
#        = - 4 * 1 * 2^3 + 2 * 1 * 2
#        = -32 + 4
#        = -28
print(f'x.grad = {x.grad!r}')

输出：

x.grad = tensor(-28.)

让我们从与模型无关的元学习算法的核心思想开始。该算法是一种与模型无关的元学习算法，它与任何使用梯度下降训练的模型兼容，并且适用于各种不同的学习问题，包括分类、回归和强化学习。元学习的目标是在各种学习任务上训练模型，以便它仅使用少量训练样本即可解决新的学习任务。

更新规则定义为：

给定微调步骤的学习率 $\alpha$ ， $\theta$ 应该最小化
$(\theta)= E _{ T _i \sim p( T )}\left[ L _{ T _i}\left(\theta_i^{\prime}\right)\right]= E _{ T _i \sim p( T )}\left[ L _{ T _i}\left(\theta-\alpha \nabla_\theta L _{ T _i}(\theta)\right)\right]$
我们首先定义一些与任务、轨迹、状态、动作和迭代相关的参数。

import argparse
from typing import NamedTupleimport gym
import numpy as np
import torch
import torch.optim as optimimport torchopt
from helpers.policy import CategoricalMLPPolicyTASK_NUM = 40
TRAJ_NUM = 20
TRAJ_LEN = 10STATE_DIM = 10
ACTION_DIM = 5GAMMA = 0.99
LAMBDA = 0.95outer_iters = 500
inner_iters = 1

接下来，我们定义一个名为 Traj 的类来表示轨迹，其中包括观察到的状态、采取的操作、采取操作后观察到的状态、获得的奖励以及用于贴现未来奖励的伽玛值。

class Traj(NamedTuple):obs: np.ndarrayacs: np.ndarraynext_obs: np.ndarrayrews: np.ndarraygammas: np.ndarray

评估函数用于评估策略在不同任务上的性能。它使用内部优化器来微调每个任务的策略，然后计算微调前后的奖励。

def evaluate(env, seed, task_num, policy):pre_reward_ls = []post_reward_ls = []inner_opt = torchopt.MetaSGD(policy, lr=0.1)env = gym.make('TabularMDP-v0',num_states=STATE_DIM,num_actions=ACTION_DIM,max_episode_steps=TRAJ_LEN,seed=args.seed,)tasks = env.sample_tasks(num_tasks=task_num)policy_state_dict = torchopt.extract_state_dict(policy)optim_state_dict = torchopt.extract_state_dict(inner_opt)for idx in range(task_num):for _ in range(inner_iters):pre_trajs = sample_traj(env, tasks[idx], policy)inner_loss = a2c_loss(pre_trajs, policy, value_coef=0.5)inner_opt.step(inner_loss)post_trajs = sample_traj(env, tasks[idx], policy)pre_reward_ls.append(np.sum(pre_trajs.rews, axis=0).mean())post_reward_ls.append(np.sum(post_trajs.rews, axis=0).mean())torchopt.recover_state_dict(policy, policy_state_dict)torchopt.recover_state_dict(inner_opt, optim_state_dict)return pre_reward_ls, post_reward_ls

在主函数中，我们初始化环境、策略和优化器。策略是一个简单的 MLP，它输出动作的分类分布。内部优化器用于在微调阶段更新策略参数，外部优化器用于在元训练阶段更新策略参数。性能通过微调前后的奖励来评估。每次外部迭代都会记录并打印训练过程。

def main(args):torch.manual_seed(args.seed)torch.cuda.manual_seed_all(args.seed)env = gym.make('TabularMDP-v0',num_states=STATE_DIM,num_actions=ACTION_DIM,max_episode_steps=TRAJ_LEN,seed=args.seed,)policy = CategoricalMLPPolicy(input_size=STATE_DIM, output_size=ACTION_DIM)inner_opt = torchopt.MetaSGD(policy, lr=0.1)outer_opt = optim.Adam(policy.parameters(), lr=1e-3)train_pre_reward = []train_post_reward = []test_pre_reward = []test_post_reward = []for i in range(outer_iters):tasks = env.sample_tasks(num_tasks=TASK_NUM)train_pre_reward_ls = []train_post_reward_ls = []outer_opt.zero_grad()policy_state_dict = torchopt.extract_state_dict(policy)optim_state_dict = torchopt.extract_state_dict(inner_opt)for idx in range(TASK_NUM):for _ in range(inner_iters):pre_trajs = sample_traj(env, tasks[idx], policy)inner_loss = a2c_loss(pre_trajs, policy, value_coef=0.5)inner_opt.step(inner_loss)post_trajs = sample_traj(env, tasks[idx], policy)outer_loss = a2c_loss(post_trajs, policy, value_coef=0.5)outer_loss.backward()torchopt.recover_state_dict(policy, policy_state_dict)torchopt.recover_state_dict(inner_opt, optim_state_dict)# Loggingtrain_pre_reward_ls.append(np.sum(pre_trajs.rews, axis=0).mean())train_post_reward_ls.append(np.sum(post_trajs.rews, axis=0).mean())outer_opt.step()test_pre_reward_ls, test_post_reward_ls = evaluate(env, args.seed, TASK_NUM, policy)train_pre_reward.append(sum(train_pre_reward_ls) / TASK_NUM)train_post_reward.append(sum(train_post_reward_ls) / TASK_NUM)test_pre_reward.append(sum(test_pre_reward_ls) / TASK_NUM)test_post_reward.append(sum(test_post_reward_ls) / TASK_NUM)print('Train_iters', i)print('train_pre_reward', sum(train_pre_reward_ls) / TASK_NUM)print('train_post_reward', sum(train_post_reward_ls) / TASK_NUM)print('test_pre_reward', sum(test_pre_reward_ls) / TASK_NUM)print('test_post_reward', sum(test_post_reward_ls) / TASK_NUM)