CARLA 深度强化学习实战：从零构建自动驾驶决策模型

1次阅读

共计 3750 个字符，预计需要花费 10 分钟才能阅读完成。

CARLA 是一款开源的自动驾驶仿真平台，它提供了高度可配置的虚拟环境，支持多种传感器模拟（如摄像头、LiDAR、雷达等），并且可以模拟复杂的交通场景和天气条件。对于自动驾驶研发来说，CARLA 是一个理想的测试平台，因为它可以快速迭代算法，而无需承担真实道路测试的高成本和风险。

传统规则式驾驶算法（如基于有限状态机的方法）虽然在特定场景下表现良好，但在复杂、动态的环境中往往显得力不从心。它们缺乏灵活性和适应性，难以处理未见的场景。相比之下，深度强化学习（DRL）通过与环境交互学习最优策略，能够更好地应对复杂多变的驾驶环境。

在连续控制任务中，常见的 DRL 算法包括 DQN、PPO 和 SAC。

DQN（Deep Q-Network）：适用于离散动作空间，但在连续控制任务中表现不佳，因为它需要对动作空间进行离散化，这会导致维度灾难。
PPO（Proximal Policy Optimization）：一种 on-policy 算法，样本效率较高，训练稳定，适合连续动作空间。
SAC（Soft Actor-Critic）：一种 off-policy 算法，结合了最大熵强化学习的优点，在探索和利用之间取得了良好的平衡，适合高维状态空间和连续动作空间。

考虑到自动驾驶任务的高维状态空间和连续动作空间，我们选择 SAC 作为基础算法，因为它具有更好的样本效率和稳定性。

为了与标准的 DRL 框架兼容，我们将 CARLA 环境封装为 gym 接口。具体实现包括：

import gym
from gym import spaces
import carla

class CarlaEnv(gym.Env):
    def __init__(self):
        super(CarlaEnv, self).__init__()
        # 定义动作空间和状态空间
        self.action_space = spaces.Box(low=-1.0, high=1.0, shape=(2,), dtype=np.float32)
        self.observation_space = spaces.Dict({"camera": spaces.Box(low=0, high=255, shape=(84, 84, 3), dtype=np.uint8),
            "lidar": spaces.Box(low=0, high=100, shape=(360,), dtype=np.float32)
        })

        # 初始化 CARLA 客户端
        self.client = carla.Client("localhost", 2000)
        self.client.set_timeout(10.0)
        self.world = self.client.get_world()

    def reset(self):
        # 重置环境
        pass

    def step(self, action):
        # 执行动作并返回新的状态、奖励、是否终止、额外信息
        pass

自动驾驶任务的状态通常包括多模态传感器数据：

摄像头图像 ：用于感知周围环境，如车道线、交通标志、其他车辆等。我们通常将图像缩放到 84×84 像素，并转换为灰度图以降低计算复杂度。
LiDAR 点云 ：用于测量障碍物的距离和方向。我们将 LiDAR 数据转换为 360 维的向量，每个维度代表一个方向的最近障碍物距离。

奖励函数的设计是 DRL 成功的关键。一个好的奖励函数应该平衡安全性、效率和舒适度：

安全性 ：碰撞惩罚、车道偏离惩罚。
效率：速度奖励、到达目标点的进度奖励。
舒适度 ：急刹车或急转弯惩罚。

def compute_reward(self, state, action):
    # 安全性
    collision_penalty = -10.0 if self.collision_detected else 0.0
    lane_penalty = -1.0 if self.lane_departure else 0.0

    # 效率
    speed_reward = 0.1 * self.current_speed
    progress_reward = 1.0 if self.reached_waypoint else 0.0

    # 舒适度
    jerk_penalty = -0.5 * np.abs(action[1])  # 刹车 / 油门
    steer_penalty = -0.2 * np.abs(action[0])  # 方向盘

    total_reward = speed_reward + progress_reward + collision_penalty + lane_penalty + jerk_penalty + steer_penalty
    return total_reward

以下是基于 PyTorch 的 SAC 实现，包含关键超参数和经验回放：

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.tensorboard import SummaryWriter
from collections import deque
import random

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

    def push(self, state, action, reward, next_state, done):
        self.buffer.append((state, action, reward, next_state, done))

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

    def __len__(self):
        return len(self.buffer)

class SAC:
    def __init__(self, state_dim, action_dim):
        # 初始化网络和优化器
        self.actor = ActorNetwork(state_dim, action_dim)
        self.critic = CriticNetwork(state_dim, action_dim)
        self.target_critic = CriticNetwork(state_dim, action_dim)

        self.actor_optimizer = optim.Adam(self.actor.parameters(), lr=3e-4)
        self.critic_optimizer = optim.Adam(self.critic.parameters(), lr=3e-4)

        # 复制参数到目标网络
        self.target_critic.load_state_dict(self.critic.state_dict())

        # 经验回放
        self.replay_buffer = ReplayBuffer(100000)

        # TensorBoard 日志
        self.writer = SummaryWriter()

    def update(self, batch_size):
        # 从经验回放中采样
        batch = self.replay_buffer.sample(batch_size)

        # 更新 critic
        # ...

        # 更新 actor
        # ...

        # 更新目标网络
        # ...

        # 记录日志
        self.writer.add_scalar("Loss/critic", critic_loss, self.steps)
        self.writer.add_scalar("Loss/actor", actor_loss, self.steps)

在训练过程中，常见的问题包括稀疏奖励和探索效率低。以下是解决方案：