一切皆是映射：深度强化元学习的挑战与机遇

    Initialize critic network Q(s,a|θ^Q) and actor network μ(s|θ^μ) with random weights
    Initialize target networks Q' and μ' with weights θ^{Q'} ← θ^Q, θ^{μ'} ← θ^μ
    Initialize replay buffer R
    for episode = 1, M do
    Initialize a random process N for action exploration
    Receive initial observation state s_1
    for t = 1, T do
        Select action a_t = μ(s_t|θ^μ) + N_t according to the current policy and exploration noise
        Execute action a_t and observe reward r_t and new state s_{t+1}
        Store transition (s_t,a_t,r_t,s_{t+1}) in R
        Sample a random minibatch of N transitions (s_i,a_i,r_i,s_{i+1}) from R
        Set y_i = r_i + γQ'(s_{i+1},μ'(s_{i+1}|θ^{μ'})|θ^{Q'})
        Update critic by minimizing the loss: L = 1/N ∑_i (y_i - Q(s_i,a_i|θ^Q))^2
        Update the actor policy using the sampled policy gradient:
            ∇_θ^μ J ≈ 1/N ∑_i ∇_a Q(s,a|θ^Q)|_{s=s_i,a=μ(s_i)} ∇_θ^μ μ(s|θ^μ)|_{s_i}
        Update the target networks:
            θ^{Q'} ← τθ^Q + (1-τ)θ^{Q'}
            θ^{μ'} ← τθ^μ + (1-τ)θ^{μ'}
    end for
    end for
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    # 定义Actor网络
    class Actor(nn.Module):
    def __init__(self, state_dim, action_dim, max_action):
        super(Actor, self).__init__()
        self.l1 = nn.Linear(state_dim, 400)
        self.l2 = nn.Linear(400, 300)
        self.l3 = nn.Linear(300, action_dim)
        self.max_action = max_action
    
    def forward(self, state):
        a = F.relu(self.l1(state))
        a = F.relu(self.l2(a))
        return self.max_action * torch.tanh(self.l3(a))
    
    # 定义Critic网络 
    class Critic(nn.Module):
    def __init__(self, state_dim, action_dim):
        super(Critic, self).__init__()
        self.l1 = nn.Linear(state_dim, 400)
        self.l2 = nn.Linear(400 + action_dim, 300)
        self.l3 = nn.Linear(300, 1)
    
    def forward(self, state, action):
        q = F.relu(self.l1(state))
        q = F.relu(self.l2(torch.cat([q, action], 1)))
        return self.l3(q)
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    class Omniglot(Dataset):
    def __init__(self, root, mode, n_way, k_shot, k_query, transform=None):
        self.root = root
        self.mode = mode
        self.n_way = n_way
        self.k_shot = k_shot
        self.k_query = k_query
        self.transform = transform
        self.img_paths, self.labels = self.load_data()
    
    def load_data(self):
        # 加载Omniglot数据集，返回图像路径和标签列表
        ...
    
    def __getitem__(self, index):
        support_set = []
        query_set = []
        classes = np.random.choice(np.unique(self.labels), self.n_way, replace=False)
        for i, cls in enumerate(classes):
            indices = np.where(self.labels == cls)[0]
            np.random.shuffle(indices)
            support_set.append(indices[:self.k_shot])
            query_set.append(indices[self.k_shot:self.k_shot+self.k_query])
        support_set = np.array(support_set).flatten()
        query_set = np.array(query_set).flatten()
    
        support_images = [self.img_paths[i] for i in support_set]
        support_labels = [self.labels[i] for i in support_set]
        query_images = [self.img_paths[i] for i in query_set]
        query_labels = [self.labels[i] for i in query_set]
    
        support_images = self.transform(default_loader(support_images))
        query_images = self.transform(default_loader(query_images))
    
        return support_images, torch.tensor(support_labels), query_images, torch.tensor(query_labels)
    
    def __len__(self):
        return len(self.img_paths)
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读

    class MAML(nn.Module):
    def __init__(self, model, inner_lr, outer_lr, n_way, inner_step):
        super(MAML, self).__init__()
        self.model = model
        self.inner_lr = inner_lr
        self.outer_lr = outer_lr
        self.n_way = n_way
        self.inner_step = inner_step
        self.loss_fn = nn.CrossEntropyLoss()
    
    def forward(self, support_set, query_set):
        fast_weights = list(self.model.parameters())
        for step in range(self.inner_step):
            support_logits = self.model.functional_forward(support_set, fast_weights)
            support_loss = self.loss_fn(support_logits, support_set[1])
            grad = torch.autograd.grad(support_loss, fast_weights)
            fast_weights = list(map(lambda p: p[1]-self.inner_lr*p[0], zip(grad, fast_weights)))
    
        query_logits = self.model.functional_forward(query_set, fast_weights)
        query_loss = self.loss_fn(query_logits, query_set[1])
    
        return query_loss
    
    def train_loop(self, epoch, train_loader, optimizer):
        print_freq = 10
        avg_loss = 0
        for i, (support_set, query_set) in enumerate(train_loader):
            loss = self.forward(support_set, query_set)
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
    
            avg_loss = avg_loss+loss.item()
    
            if i % print_freq==0:
                print('Epoch {:d} | Batch {:d}/{:d} | Loss {:f}'.format(epoch, i, len(train_loader), avg_loss/float(i+1)))
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    
    代码解读