CARLA强化学习入门实战：从零搭建自动驾驶决策模型

1次阅读

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

自动驾驶仿真训练面临几个典型挑战：

动作延迟：仿真环境与算法间的通信延迟会导致动作执行不及时，影响策略学习效果
传感器噪声：摄像头、雷达等传感器模拟数据存在噪声，增加模型鲁棒性训练难度
奖励稀疏：长序列决策任务中，正向奖励信号可能非常稀疏，导致训练效率低下

对比主流仿真平台：

CARLA 优势：
专为自动驾驶研发设计，提供高保真物理引擎
支持灵活的场景编辑和天气变化
PythonAPI 接口完善，便于强化学习集成
开源免费，社区活跃
AirSim 局限：
更侧重无人机仿真
自动驾驶相关功能相对简单
对强化学习支持不如 CARLA 直接

import carla
import numpy as np
from typing import Tuple

class CarlaEnv:
    """CARLA 强化学习环境封装"""

    def __init__(self, town: str = 'Town01'):
        self.client = carla.Client('localhost', 2000)
        self.world = self.client.load_world(town)
        self.blueprint_lib = self.world.get_blueprint_library()

    def reset(self) -> np.ndarray:
        """重置环境并返回初始观测"""
        self._destroy_actors()

        # 创建车辆和摄像头
        vehicle_bp = self.blueprint_lib.find('vehicle.tesla.model3')
        self.vehicle = self.world.spawn_actor(vehicle_bp, 
                                             random.choice(self.world.get_map().get_spawn_points()))

        camera_bp = self.blueprint_lib.find('sensor.camera.rgb')
        camera_bp.set_attribute('image_size_x', '800')
        camera_bp.set_attribute('image_size_y', '600')
        self.camera = self.world.spawn_actor(camera_bp, 
                                           carla.Transform(carla.Location(x=1.5, z=2.4)),
                                           attach_to=self.vehicle)

        # 返回处理后的图像观测
        return self._process_image(self.camera)

    def _process_image(self, image: carla.Image) -> np.ndarray:
        """将 CARLA 图像转为 numpy 数组并降采样"""
        array = np.frombuffer(image.raw_data, dtype=np.uint8)
        array = np.reshape(array, (image.height, image.width, 4))
        return array[:, :, :3]  # 去除 alpha 通道

import torch
import random
from collections import deque

class DQN(torch.nn.Module):
    """深度 Q 网络实现"""

    def __init__(self, state_dim: int, action_dim: int):
        super().__init__()
        self.net = torch.nn.Sequential(torch.nn.Conv2d(3, 32, kernel_size=8, stride=4),
            torch.nn.ReLU(),
            torch.nn.Conv2d(32, 64, kernel_size=4, stride=2),
            torch.nn.ReLU(),
            torch.nn.Flatten(),
            torch.nn.Linear(64*9*9, 512),
            torch.nn.ReLU(),
            torch.nn.Linear(512, action_dim)
        )

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        return self.net(x)

class ReplayBuffer:
    """经验回放缓冲区"""

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

    def add(self, experience: Tuple) -> None:
        self.buffer.append(experience)

    def sample(self, batch_size: int) -> Tuple:
        batch = random.sample(self.buffer, batch_size)
        states = torch.stack([x[0] for x in batch])
        actions = torch.tensor([x[1] for x in batch])
        rewards = torch.tensor([x[2] for x in batch])
        next_states = torch.stack([x[3] for x in batch])
        dones = torch.tensor([x[4] for x in batch])
        return states, actions, rewards, next_states, dones

CARLA 服务器和客户端时间不同步会导致观测和动作错位。解决方案：

使用同步模式运行服务器
在客户端设置固定时间步长
添加帧同步等待机制

# 启用同步模式
settings = world.get_settings()
settings.synchronous_mode = True
settings.fixed_delta_seconds = 0.05  # 20fps
world.apply_settings(settings)

# 在每个 step 后等待服务端更新
world.tick()

将连续动作空间离散化会导致车辆控制不连贯。改进方法：

增加离散动作数量
添加动作平滑滤波器
改用 DDPG 等连续控制算法

不同传感器数据采集时间不一致时：

使用 CARLA 的 snapshot 功能获取同一时刻数据
对滞后数据应用插值补偿
设置合理的传感器更新频率

import ray

@ray.remote
class RemoteWorker:
    def __init__(self):
        self.env = CarlaEnv()

    def sample(self, policy):
        return self.env.run_episode(policy)

# 启动多个采样 worker
workers = [RemoteWorker.remote() for _ in range(4)]
results = ray.get([w.sample.remote(policy) for w in workers])

观测数据标准化对训练稳定性至关重要：