强化学习（七）：模仿学习与逆强化学习

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

import torch
import torch.nn as nn
import numpy as np
from torch.utils.data import DataLoader, TensorDataset

class BehavioralCloning:
    """监督式模仿：最小化策略输出与专家动作之间的误差。"""

    def __init__(self, state_dim, action_dim, hidden_dims=(256, 256),
                 lr=1e-3, continuous=False, dropout=0.1):
        self.continuous = continuous
        layers, prev = [], state_dim
        for h in hidden_dims:
            layers += [nn.Linear(prev, h), nn.ReLU(), nn.Dropout(dropout)]
            prev = h
        if continuous:
            layers += [nn.Linear(prev, action_dim), nn.Tanh()]
            self.criterion = nn.MSELoss()
        else:
            layers += [nn.Linear(prev, action_dim)]
            self.criterion = nn.CrossEntropyLoss()
        self.policy = nn.Sequential(*layers)
        self.optim = torch.optim.Adam(self.policy.parameters(), lr=lr)
        self.state_mean = self.state_std = None

    def fit(self, states, actions, epochs=100, batch_size=64, val_frac=0.1):
        # 标准化状态输入——BC 是一个小模型，未归一化的特征会主导损失。
        self.state_mean = states.mean(0)
        self.state_std = states.std(0) + 1e-8
        S = (states - self.state_mean) / self.state_std

        n_val = int(len(S) * val_frac)
        idx = np.random.permutation(len(S))
        S_tr, S_val = S[idx[n_val:]], S[idx[:n_val]]
        A_tr, A_val = actions[idx[n_val:]], actions[idx[:n_val]]

        S_tr = torch.as_tensor(S_tr, dtype=torch.float32)
        S_val = torch.as_tensor(S_val, dtype=torch.float32)
        if self.continuous:
            A_tr = torch.as_tensor(A_tr, dtype=torch.float32)
            A_val = torch.as_tensor(A_val, dtype=torch.float32)
        else:
            A_tr = torch.as_tensor(A_tr, dtype=torch.long)
            A_val = torch.as_tensor(A_val, dtype=torch.long)

        loader = DataLoader(TensorDataset(S_tr, A_tr),
                            batch_size=batch_size, shuffle=True)
        best, best_state = float("inf"), None
        for epoch in range(epochs):
            self.policy.train()
            for s, a in loader:
                loss = self.criterion(self.policy(s), a)
                self.optim.zero_grad(); loss.backward(); self.optim.step()
            self.policy.eval()
            with torch.no_grad():
                v = self.criterion(self.policy(S_val), A_val).item()
            if v < best:
                best, best_state = v, {k: t.clone() for k, t in
                                       self.policy.state_dict().items()}
        self.policy.load_state_dict(best_state)

    def act(self, state):
        s = (state - self.state_mean) / self.state_std
        s = torch.as_tensor(s, dtype=torch.float32).unsqueeze(0)
        self.policy.eval()
        with torch.no_grad():
            out = self.policy(s)
        return out.squeeze().numpy() if self.continuous else out.argmax(-1).item()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

class DAgger:
    """DAgger = BC + 在学习者访问的状态上反复请专家标注。"""

    def __init__(self, state_dim, action_dim, **bc_kwargs):
        self.bc = BehavioralCloning(state_dim, action_dim, **bc_kwargs)
        self.S, self.A = [], []

    def train(self, env, expert, n_iters=10,
              n_init=50, n_per_iter=20):
        # 第 0 轮：纯专家轨迹做 warm start。
        s0, a0 = self._collect(env, expert, n_init, beta=1.0)
        self.S += s0; self.A += a0
        self.bc.fit(np.array(self.S), np.array(self.A))

        for i in range(1, n_iters + 1):
            beta = max(0.0, 1.0 - i / n_iters)   # 专家比例从 1 衰减到 0
            s, a = self._collect(env, expert, n_per_iter, beta=beta)
            self.S += s; self.A += a
            self.bc.fit(np.array(self.S), np.array(self.A))

    def _collect(self, env, expert, n_episodes, beta):
        states, actions = [], []
        for _ in range(n_episodes):
            s, done = env.reset(), False
            while not done:
                # 行动用混合策略；标签始终来自专家。
                a_play = expert(s) if np.random.rand() < beta else self.bc.act(s)
                a_label = expert(s)
                states.append(s); actions.append(a_label)
                s, _, done, _ = env.step(a_play)
        return states, actions

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

class GAIL:
    """对抗式模仿：判别器输出充当策略的每步奖励。"""

    def __init__(self, state_dim, action_dim, hidden_dim=256, continuous=False):
        self.continuous = continuous
        self.action_dim = action_dim
        in_dim = state_dim + action_dim
        self.discriminator = nn.Sequential(
            nn.Linear(in_dim, hidden_dim), nn.Tanh(),
            nn.Linear(hidden_dim, hidden_dim), nn.Tanh(),
            nn.Linear(hidden_dim, 1),
        )
        self.disc_optim = torch.optim.Adam(self.discriminator.parameters(),
                                           lr=3e-4)
        # 策略 / 价值网络：任何标准 PPO 实现都行（此处省略）。

    def _sa(self, s, a):
        if not self.continuous:
            a = torch.nn.functional.one_hot(a.long(), self.action_dim).float()
        return torch.cat([s, a], dim=-1)

    def reward(self, states, actions):
        """喂给策略学习器的每步模仿奖励。"""
        with torch.no_grad():
            logits = self.discriminator(self._sa(states, actions))
            d = torch.sigmoid(logits)
            r = -torch.log(1.0 - d + 1e-8)   # 原版 GAIL 的奖励形式
        return r.squeeze(-1)

    def update_discriminator(self, expert_sa, policy_sa):
        # 二元交叉熵：专家=0，策略=1。
        e_logits = self.discriminator(expert_sa)
        p_logits = self.discriminator(policy_sa)
        bce = torch.nn.functional.binary_cross_entropy_with_logits
        loss = bce(e_logits, torch.zeros_like(e_logits)) + \
               bce(p_logits, torch.ones_like(p_logits))
        self.disc_optim.zero_grad(); loss.backward(); self.disc_optim.step()
        return loss.item()