第11章：端到端自动驾驶

传统模块化架构：
┌────────┐   ┌────────┐   ┌────────┐   ┌────────┐
│Sensors │→ │Percep- │→ │Predic- │→ │Planning│→ Control
│        │   │tion    │   │tion    │   │        │
└────────┘   └────────┘   └────────┘   └────────┘
    ↓            ↓            ↓            ↓
Raw Data    3D Boxes    Trajectories  Waypoints

端到端架构：
┌────────┐   ┌─────────────────────────┐   ┌────────┐
│Sensors │→ │   End-to-End Network    │→ │Control │
│        │   │  (Perception+Planning)  │   │        │
└────────┘   └─────────────────────────┘   └────────┘
    ↓                    ↓                      ↓
Raw Data         Learned Features          Steering/Acc

L_imitation = E[(a_expert - π(x, g))²]  # 行为克隆
L_RL = -E[Σ γᵗ r(sₜ, aₜ)]              # 强化学习

          Multi-Camera Images
                  ↓
         ┌─────────────────┐
         │  BEV Encoder    │ (Deformable Attention)
         └─────────────────┘
                  ↓
         ┌─────────────────┐
         │ Query Generator │ → Track/Map/Agent/Goal Queries
         └─────────────────┘
                  ↓
    ┌──────────────────────────┐
    │   Transformer Decoder     │
    │  ┌────────────────────┐  │
    │  │ Perception Head    │  │ → 3D Detection
    │  ├────────────────────┤  │
    │  │ Prediction Head    │  │ → Trajectories  
    │  ├────────────────────┤  │
    │  │ Planning Head      │  │ → Ego Trajectory
    │  └────────────────────┘  │
    └──────────────────────────┘

Vision Encoder → Vision-Language Alignment → Driving Policy
     ↓                     ↓                       ↓
Camera Input     Natural Language          Control Output
                   Description

优势：

- 可解释性强：可以检查感知结果是否正确
- 便于调试：可以定位问题出现在哪个阶段  
- 知识迁移：可以利用现有感知数据集预训练

劣势：

- 表示瓶颈：预定义的表示可能丢失信息
- 标注成本高：需要详细的中间标注
- 优化困难：多个损失函数的权重平衡

优势：

- 信息保真：保留所有可能有用的信息
- 端到端优化：特征完全为最终任务服务
- 灵活适应：可以学习难以手工定义的模式

劣势：

- 黑箱特性：难以理解网络学到了什么
- 调试困难：错误难以定位和修复
- 泛化风险：可能过拟合训练分布

# 伪代码示例
class HybridE2EModel:
    def forward(self, sensor_input, supervision_level='full'):
        # 共享的特征提取
        features = self.backbone(sensor_input)

        # 可选的显式感知输出
        if supervision_level in ['full', 'perception']:
            detection = self.detection_head(features)
            segmentation = self.segmentation_head(features)
            losses['perception'] = compute_perception_loss(...)

        # 隐式规划特征
        planning_features = self.planning_encoder(features)

        # 最终控制输出
        control = self.control_head(planning_features)
        losses['control'] = compute_control_loss(...)

        return control, losses

Cross-Attention可视化示例：
Query: 未来3秒的轨迹点
Key/Value: 图像特征

┌─────────────────────────┐
│  前视摄像头图像         │
│  ┌───┬───┬───┬───┐     │
│  │░░░│███│███│░░░│     │ ← 高注意力区域（行人过街）
│  ├───┼───┼───┼───┤     │
│  │░░░│▓▓▓│▓▓▓│░░░│     │ ← 中等注意力（车道线）
│  ├───┼───┼───┼───┤     │
│  │░░░│░░░│░░░│░░░│     │ ← 低注意力（天空）
│  └───┴───┴───┴───┘     │
└─────────────────────────┘

训练阶段特征演化：

早期（<10k步）：

- 学习基本的视觉特征（边缘、颜色）
- 简单的运动模式（直行、停止）

中期（10k-100k步）：

- 物体级别的理解（车辆、行人）
- 基本的交通规则（红灯停、绿灯行）

后期（>100k步）：

- 场景级别的理解（路口、高速公路）
- 复杂的交互行为（礼让、汇入）

def analyze_gradient_flow(model, loss):
    """分析梯度流以诊断训练问题"""
    loss.backward()

    for name, param in model.named_parameters():
        if param.grad is not None:
            grad_norm = param.grad.norm().item()

            # 检查梯度消失
            if grad_norm < 1e-5:
                print(f"梯度消失: {name}")

            # 检查梯度爆炸
            elif grad_norm > 1e3:
                print(f"梯度爆炸: {name}")

1. 真实驾驶数据：
   车队采集 → 数据清洗 → 场景提取 → 质量筛选

2. 困难样本挖掘：
   部署模型 → 识别失败案例 → 定向采集 → 重新训练

3. 仿真数据生成：
   场景参数化 → 物理仿真 → 传感器模拟 → 数据渲染

# 行为克隆损失函数
def behavior_cloning_loss(predictions, expert_actions, weights=None):
    """
    predictions: 模型预测的动作
    expert_actions: 专家（人类）动作
    weights: 样本权重（可选）
    """
    # L2损失用于连续控制
    l2_loss = F.mse_loss(predictions, expert_actions, reduction='none')

    if weights is not None:
        l2_loss = l2_loss * weights

    return l2_loss.mean()

迭代过程：

1. 用当前策略收集数据
2. 专家对新状态提供标签
3. 将新数据加入训练集
4. 重新训练策略

def cql_loss(q_values, actions, rewards, next_q_values, alpha=1.0):
    """
    保守Q学习损失，惩罚对未见动作的过高估值
    """
    # 标准TD损失
    td_loss = F.mse_loss(
        q_values, 
        rewards + gamma * next_q_values.max(dim=1)[0]
    )

    # CQL正则化项：降低OOD动作的Q值
    cql_reg = torch.logsumexp(q_values, dim=1).mean() - q_values.mean()

    return td_loss + alpha * cql_reg

网络结构：
┌─────────┐
│State    │ → Value Network → V(s)
│Encoder  │ → Q Network → Q(s,a)
└─────────┘ → Policy Network → π(a|s)

训练目标：

- Value: 回归到Q值的期望分位数
- Q-function: 标准TD学习
- Policy: 通过优势加权的行为克隆

# 域随机化参数示例
randomization_params = {
    'lighting': {
        'sun_angle': [0, 360],
        'intensity': [0.3, 1.5],
        'shadow_strength': [0, 1]
    },
    'weather': {
        'rain_intensity': [0, 1],
        'fog_density': [0, 0.5],
        'wetness': [0, 1]
    },
    'sensor': {
        'camera_fov': [60, 120],
        'noise_level': [0, 0.1],
        'exposure': [0.5, 2.0]
    },
    'dynamics': {
        'friction': [0.5, 1.5],
        'mass_variation': [0.8, 1.2],
        'actuator_delay': [0, 100]  # ms
    }
}

训练策略：

1. 仿真预训练：在高保真仿真器中训练基础模型
2. 混合微调：真实数据 + 仿真数据联合训练
3. 真实数据精调：仅使用真实数据微调关键层
4. 在线适应：部署后持续学习新场景

架构：
         ┌──────────────┐
Sim Data → Feature      → Task Head → Control
         │  Encoder     │
Real Data→              → Domain Disc → Real/Sim
         └──────────────┘

损失函数：
L_total = L_task + λ_adv * L_adversarial + λ_grad * L_gradient_reversal

# 神经场景生成pipeline
class NeuralSceneGenerator:
    def __init__(self):
        self.nerf_model = load_pretrained_nerf()
        self.asset_library = load_3d_assets()

    def generate_scene(self, scenario_params):
        # 1. 生成基础场景布局
        layout = self.generate_layout(scenario_params)

        # 2. 放置动态物体
        objects = self.place_objects(layout, scenario_params)

        # 3. 神经渲染生成图像
        images = self.nerf_render(layout, objects)

        # 4. 添加传感器噪声
        noisy_images = self.add_sensor_effects(images)

        return noisy_images

# SCENIC场景描述示例
scenario Main():
    # 定义道路网络
    road = Road.straight(length=100m)

    # 自车配置
    ego = Car at road.start

    # 其他车辆
    leadCar = Car at (ego + (20m, 0)),
              with speed in [10, 20] m/s

    # 行人行为
    pedestrian = Pedestrian at road.sidewalk,
                  with behavior CrossingBehavior()

    # 约束条件
    require distance(ego, leadCar) > 5m
    require visibility(ego, pedestrian) > 0.8

class ShadowModeValidator:
    def __init__(self, model, safety_threshold=0.1):
        self.model = model
        self.safety_threshold = safety_threshold
        self.divergence_events = []

    def validate(self, sensor_data, human_action):
        # 模型预测
        model_action = self.model(sensor_data)

        # 计算偏差
        divergence = self.compute_divergence(model_action, human_action)

        # 记录显著偏差
        if divergence > self.safety_threshold:
            self.divergence_events.append({
                'timestamp': time.now(),
                'sensor_data': sensor_data,
                'model_action': model_action,
                'human_action': human_action,
                'divergence': divergence
            })

        return divergence

部署阶段：

1. 仿真验证（>1M英里）
2. 封闭场地测试（结构化场景）
3. 限定ODD测试（如高速公路）
4. 扩展ODD测试（城市道路）
5. 影子模式大规模验证
6. 有安全员的道路测试
7. 无安全员部署（地理围栏区域）

def reachability_analysis(model, input_bounds):
    """
    使用区间传播分析模型输出范围
    """
    # 将输入边界转换为区间
    interval = Interval(input_bounds['min'], input_bounds['max'])

    # 通过网络传播区间
    for layer in model.layers:
        if isinstance(layer, Linear):
            interval = interval.linear_transform(layer.weight, layer.bias)
        elif isinstance(layer, ReLU):
            interval = interval.relu()

    return interval.bounds()

def verify_robustness(model, x, epsilon, property_fn):
    """
    验证模型在epsilon扰动下是否满足安全属性
    """
    # 定义输入扰动范围
    input_bounds = {
        'min': x - epsilon,
        'max': x + epsilon
    }

    # 计算输出范围
    output_bounds = reachability_analysis(model, input_bounds)

    # 检查是否满足属性
    return property_fn(output_bounds)

class UncertaintyAwareE2E:
    def __init__(self, models):
        self.ensemble = models  # 多个模型的集成

    def predict_with_uncertainty(self, x):
        predictions = []
        for model in self.ensemble:
            predictions.append(model(x))

        predictions = torch.stack(predictions)
        mean = predictions.mean(dim=0)
        std = predictions.std(dim=0)

        # 认知不确定性（模型不确定性）
        epistemic = std

        # 总不确定性
        uncertainty = epistemic

        return mean, uncertainty

class AnomalyDetector:
    def __init__(self, threshold=3.0):
        self.threshold = threshold
        self.feature_stats = None

    def fit(self, normal_features):
        """在正常驾驶数据上训练"""
        self.mean = normal_features.mean(dim=0)
        self.std = normal_features.std(dim=0)

    def detect(self, features):
        """检测异常输入"""
        z_score = (features - self.mean) / (self.std + 1e-6)
        anomaly_score = z_score.abs().max()

        is_anomaly = anomaly_score > self.threshold
        return is_anomaly, anomaly_score

安全包络层次：
┌─────────────────────────────┐
│   端到端神经网络控制器      │ ← 主控制器
├─────────────────────────────┤
│   规则基础安全检查器        │ ← 过滤危险动作
├─────────────────────────────┤
│   紧急制动系统（AEB）       │ ← 最后防线
└─────────────────────────────┘

class SafetyEnvelope:
    def __init__(self, e2e_model, fallback_controller):
        self.e2e_model = e2e_model
        self.fallback = fallback_controller
        self.safety_checker = SafetyChecker()

    def control(self, state):
        # 1. 端到端模型预测
        e2e_action, confidence = self.e2e_model.predict_with_confidence(state)

        # 2. 安全性检查
        is_safe = self.safety_checker.check(state, e2e_action)

        # 3. 降级决策
        if not is_safe or confidence < 0.8:
            # 切换到保守的规则控制器
            action = self.fallback.control(state)
            mode = "fallback"
        else:
            action = e2e_action
            mode = "e2e"

        # 4. 最终安全约束
        action = self.apply_safety_constraints(action, state)

        return action, mode

降级层次：

1. 完全自动驾驶（L4）
   ↓ (系统不确定性高)

2. 限制场景自动驾驶（仅高速）
   ↓ (传感器故障)

3. 辅助驾驶（L2）
   ↓ (多传感器故障)

4. 最小风险机动（靠边停车）