第15章：行为克隆与模仿学习

图像输入 → CNN特征提取器 → 
         ↓
    空间注意力机制
         ↓
    特征融合层 → 策略头

视觉流: ResNet/ViT → 视觉特征
                    ↘
                     融合模块 → 共享表示 → 动作预测
                    ↗
本体感知流: MLP → 本体特征

主臂(人操作) → 运动捕捉 → 运动重定向 → 从臂(机器人)
                           ↓
                      数据记录系统

# 伪代码：多模态数据同步
class DataRecorder:
    def __init__(self):
        self.time_offset = calibrate_clocks()
        self.buffer = CircularBuffer(size=1000)

    def record_step(self):
        t = get_synchronized_time()
        data = {
            'timestamp': t,
            'rgb': camera.get_frame(t),
            'depth': depth_camera.get_frame(t),
            'joint_pos': robot.get_joints(t),
            'joint_vel': robot.get_velocities(t),
            'ee_pose': robot.get_ee_pose(t),
            'action': teleop.get_action(t),
            'force_torque': ft_sensor.get_reading(t)
        }
        self.buffer.add(data)

视觉域随机化参数：

- 光照：强度[0.5, 2.0]，色温[4000K, 7000K]
- 纹理：随机替换物体和背景纹理
- 相机：内参扰动±5%，位置噪声±5cm
- 物理：摩擦系数[0.5, 1.5]，质量±20%

专家轨迹：  s₀ → s₁ → s₂ → s₃ → ... → sₜ
            ↓    ↓    ↓    ↓         ↓
专家动作：  a₀*  a₁*  a₂*  a₃*  ...  aₜ*

学习轨迹：  s₀ → s₁' → s₂' → s₃' → ... → sₜ'
            ↓    ↓     ↓     ↓          ↓
预测动作：  â₀   â₁    â₂    â₃   ...   âₜ

偏差累积：  0 → δ₁ → δ₁+δ₂ → δ₁+δ₂+δ₃ → ...

算法: DAgger
输入: 初始数据集 D₀ = {(s,a*)}从专家收集
      专家策略 π*
      迭代次数 N

1. 在 D₀ 上训练初始策略 π₁
2. for i = 1 to N:
   a. 用当前策略 πᵢ 收集轨迹
   b. 对轨迹中的每个状态 s，查询专家动作 a* = π*(s)
   c. 聚合数据：Dᵢ = Dᵢ₋₁ ∪ {(s, a*)}
   d. 在 Dᵢ 上重新训练策略 πᵢ₊₁

3. 返回最终策略 πₙ₊₁

def safe_dagger_step(state, learned_policy, expert_policy, safety_checker):
    if safety_checker.is_safe(state):
        action = learned_policy(state)
        label = expert_policy(state)  # 仍然记录专家标签
    else:
        action = expert_policy(state)  # 专家接管
        label = action
    return action, (state, label)

class EnsembleDAgger:
    def __init__(self, n_models=5):
        self.models = [create_model() for _ in range(n_models)]

    def predict(self, state):
        predictions = [m(state) for m in self.models]
        action = np.mean(predictions, axis=0)
        uncertainty = np.std(predictions, axis=0)
        return action, uncertainty

    def should_query_expert(self, uncertainty, threshold=0.1):
        return np.max(uncertainty) > threshold

1. 初始化奖励函数 R
2. 重复：
   a. 用当前R求解最优策略 π
   b. 计算特征期望：
      μ_π = E[Σ γ^t φ(s_t, a_t) | π]
      μ_E = E[Σ γ^t φ(s_t, a_t) | 专家演示]
   c. 更新奖励函数使 μ_π 接近 μ_E

3. 返回学习到的奖励函数 R

def soft_value_iteration(R, gamma=0.99, threshold=1e-4):
    V = np.zeros(n_states)
    while True:
        V_new = soft_bellman_backup(V, R, gamma)
        if np.max(np.abs(V - V_new)) < threshold:
            break
        V = V_new

    # 计算软Q函数
    Q = R + gamma * expected_V(V)

    # 导出策略
    π = np.exp(Q - logsumexp(Q, axis=1, keepdims=True))
    return π, V, Q

def soft_bellman_backup(V, R, gamma):
    Q = R + gamma * V_next  # V_next是下一状态的值
    return logsumexp(Q, axis=1)  # 软最大值

class NeuralRewardFunction(nn.Module):
    def __init__(self, state_dim, action_dim, hidden_dims=[256, 256]):
        super().__init__()
        input_dim = state_dim + action_dim
        layers = []
        for h_dim in hidden_dims:
            layers.extend([
                nn.Linear(input_dim, h_dim),
                nn.ReLU(),
                nn.Dropout(0.1)
            ])
            input_dim = h_dim
        layers.append(nn.Linear(input_dim, 1))
        self.net = nn.Sequential(*layers)

    def forward(self, states, actions):
        x = torch.cat([states, actions], dim=-1)
        return self.net(x)

传统IRL+RL: 专家演示 → 奖励函数 → 最优策略
GAIL:       专家演示 → 最优策略 (通过对抗学习)

class Discriminator(nn.Module):
    def __init__(self, state_dim, action_dim, hidden_dims=[256, 256]):
        super().__init__()
        input_dim = state_dim + action_dim
        layers = []
        for h_dim in hidden_dims:
            layers.extend([
                nn.Linear(input_dim, h_dim),
                nn.Tanh(),  # Tanh通常比ReLU更稳定
            ])
            input_dim = h_dim
        layers.append(nn.Linear(input_dim, 1))
        self.net = nn.Sequential(*layers)

    def forward(self, states, actions):
        return torch.sigmoid(self.net(torch.cat([states, actions], dim=-1)))

    def compute_reward(self, states, actions):
        # 用判别器输出作为奖励信号
        D = self.forward(states, actions)
        return torch.log(D + 1e-8) - torch.log(1 - D + 1e-8)

class GAILTrainer:
    def __init__(self, policy, discriminator, expert_dataset):
        self.policy = policy
        self.discriminator = discriminator
        self.expert_dataset = expert_dataset
        self.policy_optimizer = PPO(policy)

    def train_step(self):
        # 1. 收集策略轨迹
        policy_batch = self.collect_trajectories(self.policy)

        # 2. 采样专家数据
        expert_batch = self.expert_dataset.sample(len(policy_batch))

        # 3. 更新判别器
        self.update_discriminator(expert_batch, policy_batch)

        # 4. 计算策略奖励
        rewards = self.discriminator.compute_reward(
            policy_batch.states, 
            policy_batch.actions
        )

        # 5. 更新策略
        self.policy_optimizer.update(policy_batch, rewards)

from torch.nn.utils import spectral_norm

class SpectralNormDiscriminator(nn.Module):
    def __init__(self, state_dim, action_dim):
        super().__init__()
        self.net = nn.Sequential(
            spectral_norm(nn.Linear(state_dim + action_dim, 256)),
            nn.Tanh(),
            spectral_norm(nn.Linear(256, 256)),
            nn.Tanh(),
            spectral_norm(nn.Linear(256, 1))
        )

class ReplayBuffer:
    def __init__(self, capacity=10000):
        self.buffer = deque(maxlen=capacity)

    def add(self, trajectories):
        self.buffer.extend(trajectories)

    def sample(self, batch_size):
        return random.sample(self.buffer, batch_size)

# 判别器学习率应该较小
d_lr = 3e-4
g_lr = 3e-4

# 使用不同的更新频率
d_updates_per_g_update = 5  # 判别器更新更频繁

传感器输入 → 感知网络 → 特征融合 → 策略网络 → 控制输出
     ↑                      ↑
  8个相机              BEV特征空间

# 伪代码：多任务策略网络
class FSDPolicyNetwork(nn.Module):
    def __init__(self):
        self.backbone = VisionTransformer()
        self.bev_decoder = BEVDecoder()

        # 多个输出头
        self.trajectory_head = TrajectoryHead()
        self.speed_head = SpeedHead()  
        self.lane_change_head = LaneChangeHead()
        self.traffic_light_head = TrafficLightHead()

    def forward(self, camera_inputs):
        features = self.backbone(camera_inputs)
        bev_features = self.bev_decoder(features)

        outputs = {
            'trajectory': self.trajectory_head(bev_features),
            'speed': self.speed_head(bev_features),
            'lane_change': self.lane_change_head(bev_features),
            'traffic_light': self.traffic_light_head(bev_features)
        }
        return outputs

部署 → 收集困难案例 → 人工标注 → 重训练 → 验证 → 部署
         ↑                                    ↓
         ←────────────────────────────────────

class CQL:
    def __init__(self, q_network, policy, alpha=1.0):
        self.q_network = q_network
        self.policy = policy
        self.alpha = alpha

    def compute_cql_loss(self, states, actions, rewards, next_states, dones):
        # 标准TD损失
        q_values = self.q_network(states, actions)
        with torch.no_grad():
            next_q = self.compute_target_q(next_states)
            target_q = rewards + (1 - dones) * self.gamma * next_q
        td_loss = F.mse_loss(q_values, target_q)

        # CQL正则化项
        # 计算log-sum-exp over actions
        num_samples = 10
        random_actions = torch.rand(states.shape[0], num_samples, self.action_dim)
        random_q = self.q_network(states.unsqueeze(1), random_actions)
        logsumexp_q = torch.logsumexp(random_q, dim=1)

        # 数据集动作的Q值
        dataset_q = self.q_network(states, actions)

        # CQL损失
        cql_loss = (logsumexp_q - dataset_q).mean()

        return td_loss + self.alpha * cql_loss

def iql_loss(q, v, states, actions, rewards, next_states):
    # 学习V函数(期望Q值的分位数)
    with torch.no_grad():
        target_q = q(states, actions)
    v_loss = expectile_loss(v(states), target_q, tau=0.7)

    # 学习Q函数
    with torch.no_grad():
        next_v = v(next_states)
    target = rewards + gamma * next_v
    q_loss = F.mse_loss(q(states, actions), target)

    # 通过advantage加权行为克隆学习策略
    with torch.no_grad():
        advantage = q(states, actions) - v(states)
        weight = torch.exp(advantage / beta)
    policy_loss = -weight * log_prob(actions).mean()

    return q_loss + v_loss + policy_loss

def assess_dataset_quality(dataset):
    metrics = {
        'coverage': compute_state_coverage(dataset),
        'return_distribution': compute_return_stats(dataset),
        'trajectory_length': compute_trajectory_lengths(dataset),
        'action_diversity': compute_action_entropy(dataset)
    }
    return metrics

# 错误：偏向性数据收集
if task_successful:
    dataset.add(trajectory)  # 只记录成功案例

# 正确：全面数据收集
dataset.add(trajectory)
if not task_successful:
    dataset.add_recovery_demo()  # 添加恢复演示

# 错误：忽略历史
action = policy(current_state)

# 正确：包含历史
action = policy(current_state, history[-k:])

# 错误：忽视分布偏移
model.eval()
deploy_to_robot(model)  # 危险！

# 正确：安全部署
for confidence_threshold in [0.9, 0.7, 0.5]:
    deploy_with_safety_net(model, confidence_threshold)
    collect_failure_cases()
    retrain_model()

# 错误：不adaptive
d_optimizer = Adam(lr=1e-3)
g_optimizer = Adam(lr=1e-3)

# 正确：动态调整
if discriminator_accuracy > 0.8:
    d_lr *= 0.5  # 降低判别器学习率
if generator_loss > threshold:
    g_lr *= 1.1  # 提高生成器学习率

# 错误：Q值过估计
action = argmax(Q(state, a) for a in action_space)

# 正确：CQL保守策略
action = sample_from_policy(state)
if Q(state, action) < conservative_threshold:
    action = closest_dataset_action(state)