CARLA强化学习实战：从环境搭建到自动驾驶策略训练

1次阅读

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

CARLA 作为开源的自动驾驶仿真平台，相比 AirSim 具有三个核心优势：

真实感地图数据 ：基于 OpenDRIVE 标准的路网描述，支持 Town01-Town07 等多种复杂城市场景
灵活的传感器配置 ：可自由组合摄像头（RGB/Depth/Semantic）、激光雷达、雷达等传感器
完善的 PythonAPI：提供从车辆控制到交通流管理的全套接口，特别适合强化学习智能体开发

与 AirSim 相比，CARLA 在物理引擎精度（使用 Unreal Engine）、交通参与者多样性（行人 / 自行车 / 动态天气）方面表现更优，但需要更高配置的硬件支持。

推荐使用 Docker 镜像部署服务端：

docker pull carlasim/carla:0.9.13

关键配置参数：

-carla-rpc-port=2000 设置通信端口
-quality-level=Epic 图形质量（Low/Epic）
-benchmark -fps=20 固定帧率模式

# requirements.txt
carla==0.9.13
numpy>=1.21
torch==1.12.1+cu113  # 需与 CUDA 版本匹配
gym==0.26.2

class CameraProcessor:
    def __init__(self, width=800, height=600):
        self.transform = T.Compose([T.ToPILImage(),
            T.Resize((256, 256)),  # 统一输入尺寸
            T.ToTensor(),
            T.Normalize(mean=[0.485, 0.456, 0.406], 
                       std=[0.229, 0.224, 0.225])  # ImageNet 统计量
        ])

    def process(self, carla_image):
        array = np.frombuffer(carla_image.raw_data, dtype=np.uint8)
        array = array.reshape((carla_image.height, carla_image.width, 4))[:,:,:3]
        return self.transform(array)

def lidar_to_voxel(points, voxel_size=0.1, z_range=(-2,5)):
    # points: (N,3) numpy 数组
    voxel_grid = np.zeros((int(100/voxel_size), int(100/voxel_size), int((z_range[1]-z_range[0])/voxel_size))
    )

    # 过滤高度范围外的点
    mask = (points[:,2] >= z_range[0]) & (points[:,2] <= z_range[1])
    points = points[mask]

    # 转换为体素坐标
    indices = ((points[:,:2] + 50) / voxel_size).astype(int)
    z_indices = ((points[:,2] - z_range[0]) / voxel_size).astype(int)

    # 避免越界
    valid = (indices[:,0] >= 0) & (indices[:,0] < voxel_grid.shape[0]) \
          & (indices[:,1] >= 0) & (indices[:,1] < voxel_grid.shape[1])

    # 占据标记
    voxel_grid[indices[valid,0], indices[valid,1], z_indices[valid]] = 1
    return voxel_grid

def calculate_reward(vehicle, prev_state, action):
    # 安全指标
    collision = vehicle.get_collision_history()
    safety_reward = -10.0 if len(collision) > 0 else 0.0

    # 效率指标
    current_speed = get_speed(vehicle)
    target_speed = 50 * 0.27778  # km/h -> m/s
    speed_reward = 1 - abs(current_speed - target_speed)/target_speed

    # 舒适度指标
    jerk = np.linalg.norm(action - prev_state["last_action"])
    comfort_reward = -0.1 * jerk

    # 车道保持
    lane_offset = abs(vehicle.get_transform().location.y - road_center_y)
    lane_reward = -0.5 * (lane_offset / 3.0)  # 3m 为车道半宽

    return safety_reward * 0.6 + speed_reward * 0.3 + comfort_reward * 0.05 + lane_reward * 0.05

class PPOTrainer:
    def __init__(self, policy, clip_param=0.2, lr=3e-4):
        self.policy = policy
        self.optimizer = torch.optim.Adam(policy.parameters(), lr=lr)

    def update(self, samples):
        states, actions, old_log_probs, returns, advantages = samples

        # 重要性采样
        new_dist = self.policy.get_distribution(states)
        new_log_probs = new_dist.log_prob(actions)

        ratio = (new_log_probs - old_log_probs).exp()
        surr1 = ratio * advantages
        surr2 = torch.clamp(ratio, 1.0 - self.clip_param, 1.0 + self.clip_param) * advantages

        policy_loss = -torch.min(surr1, surr2).mean()
        value_loss = 0.5 * (returns - self.policy.value(states)).pow(2).mean()

        entropy_loss = new_dist.entropy().mean()  # 鼓励探索

        self.optimizer.zero_grad()
        (policy_loss + 0.5 * value_loss - 0.01 * entropy_loss).backward()
        torch.nn.utils.clip_grad_norm_(self.policy.parameters(), 0.5)
        self.optimizer.step()

Batch Size 选择 ：
输入尺寸 256×256 时，建议 batch=32（16GB 显存）
使用梯度累积（accumulation_steps=4）模拟更大 batch

网络深度权衡 ：

# ResNet18 vs ResNet34 实测对比 (FPS @ RTX3090)
# Backbone | GPU Mem(MB) | FPS
# ResNet18 | 3421        | 58
# ResNet34 | 4895        | 43

class ParallelEnv:
    def __init__(self, num_workers=4):
        self.workers = []
        self.pipes = []

        for _ in range(num_workers):
            parent_conn, child_conn = mp.Pipe()
            p = mp.Process(target=worker_process, args=(child_conn,))
            p.start()
            self.workers.append(p)
            self.pipes.append(parent_conn)

    def step(self, actions):
        for pipe, action in zip(self.pipes, actions):
            pipe.send(("step", action))

        return [pipe.recv() for pipe in self.pipes]

当动作空间离散化时（如转向分 10 档 + 油门分 5 档），会产生 50 维组合。解决方案：

使用连续动作空间 + Beta 分布（范围受限）
层次化决策：高层输出宏观指令，底层执行具体控制

域随机化（Domain Randomization）：

# 在环境重置时随机化这些参数
def reset(self):
    weather_params = {"clouds": random.uniform(0,100),
        "rain": random.uniform(0,100),
        "sun_altitude": random.uniform(15,90)
    }
    self.world.set_weather(weather_params)

添加传感器噪声模型：

def add_noise(image):
    # 高斯噪声
    noise = torch.randn_like(image) * 0.05
    # 运动模糊
    if random.random() < 0.3:
        kernel_size = random.choice([3,5,7])
        image = T.functional.gaussian_blur(image, kernel_size)
    return torch.clamp(image + noise, 0, 1)

高层策略 ：处理路线规划、变道决策等长周期任务（1Hz 更新）
中层策略 ：应对交通灯、行人避让等场景（5Hz 更新）
底层控制器 ：执行转向 / 油门等即时控制（20Hz 更新）

安全层设计 ：在神经网络输出后添加规则校验

def safety_filter(action):
    steer, throttle = action
    if abs(steer) > 0.8 and throttle > 0.5:  # 高速急转弯
        return np.clip(steer, -0.5, 0.5), 0.3
    return action