基于CARLA仿真平台的深度强化学习实战：从算法设计到避坑指南

1次阅读

没有评论

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

自动驾驶仿真训练中，开发者常遇到几个典型问题：

奖励稀疏性：车辆在复杂场景中长时间得不到有效反馈，导致策略更新缓慢
环境随机性：行人、车辆等动态物体的随机行为使得训练难以收敛
训练效率低：高保真物理引擎带来的计算开销导致迭代周期长
策略脆弱性：在仿真环境表现良好的策略，迁移到真实世界时出现性能下降

在 CARLA 环境中对比主流 DRL 算法表现：

DQN：
适合离散动作空间
难以处理连续控制任务（如方向盘转角）
样本效率较低
SAC：
对超参数敏感
训练初期探索成本高
需要精细的温度系数调节
PPO：
支持连续动作空间
通过 clip 机制保证策略更新稳定性
样本利用率较高（可复用旧样本）
超参数鲁棒性较好

关键选择依据：PPO 的 clip 机制（ε 通常取 0.1-0.3）能有效防止 CARLA 环境中因奖励函数设计不当导致的策略崩溃。

class CarlaEnv(gym.Env):
    def __init__(self, town="Town01", render_mode="no_render"):
        # 初始化客户端
        self.client = carla.Client("localhost", 2000)
        self.client.set_timeout(10.0)

        # 关键配置
        self.world = self.client.load_world(town)
        self.settings = self.world.get_settings()
        self.settings.synchronous_mode = True  # 必须启用同步模式
        self.settings.fixed_delta_seconds = 0.05  # 20FPS

        # 传感器配置示例
        self.camera = CameraSetup(
            image_size_x=640,
            image_size_y=320,
            fov=90,
            transform=carla.Transform(carla.Location(z=2.4))
        )

设计多目标加权奖励函数：

车道保持奖励（权重 0.3）：

def _lane_reward(self):
    waypoint = self.map.get_waypoint(self.vehicle.get_location())
    distance = waypoint.transform.location.distance(self.vehicle.get_location())
    return 1.0 - min(distance / 3.0, 1.0)  # 归一化到[0,1]

碰撞惩罚（权重 -1.0）：

def _collision_penalty(self):
    if self.collision_sensor.has_collided:
        return -1.0 * min(self.collision_sensor.intensity, 10.0)
    return 0.0

速度奖励（权重 0.2）：

def _speed_reward(self):
    target_speed = 50  # km/h
    current_speed = 3.6 * math.sqrt(self.velocity.x**2 + self.velocity.y**2)
    return max(0, 1 - abs(current_speed - target_speed)/target_speed)

使用 Stable Baselines3 的优化实现：

from stable_baselines3 import PPO
from stable_baselines3.common.callbacks import CheckpointCallback

# 关键参数配置
model = PPO(
    policy="CnnPolicy",
    env=train_env,
    learning_rate=3e-4,  # 建议范围[1e-5, 1e-3]
    n_steps=2048,       # 与 CARLA 的同步步长匹配
    batch_size=64,      # 显存不足时可减小
    n_epochs=10,        # 建议 5 -15 次
    gamma=0.99,         # 折扣因子
    gae_lambda=0.95,    # GAE 参数
    ent_coef=0.01,      # 熵正则化系数
    verbose=1
)

# 训练配置
checkpoint_callback = CheckpointCallback(
    save_freq=100000,
    save_path="./logs/"
)
model.learn(
    total_timesteps=1e6,
    callback=checkpoint_callback,
    log_interval=5
)

主节点：运行 PPO 算法更新
Worker 节点：
每个 worker 运行独立 CARLA 实例
通过 Ray 框架实现并行采样
参数服务器：
集中管理策略参数
使用 NCCL 加速通信

在 CARLA 启动时添加参数：

./CarlaUE4.sh -quality-level=Low -nosound -RenderOffScreen -benchmark -fps=20

吞吐量对比：

模式	FPS	GPU 显存占用
默认渲染	15	8GB
NoRender	45	3GB

import torch

def print_gpu_usage():
    allocated = torch.cuda.memory_allocated() / 1024**3
    reserved = torch.cuda.memory_reserved() / 1024**3
    print(f"Allocated: {allocated:.2f}GB, Reserved: {reserved:.2f}GB")

定期重启 CARLA 服务（每 2 小时）

使用上下文管理器管理连接：

with CarlaConnection() as client:
    # 业务代码

检查 PythonAPI 版本匹配（推荐 0.9.12+）

经验值范围：

单步奖励建议控制在[-1,1]
回合总奖励建议在 [-10,10] 区间

使用 RunningMeanStd 进行自动归一化：

from stable_baselines3.common.running_mean_std import RunningMeanStd
self.returns_rms = RunningMeanStd(shape=1)

使用 Queue 进行进程间通信

对共享变量加锁：

from threading import Lock
self._lock = Lock()

def update_state(self, value):
    with self._lock:
        self.shared_state = value

def calculate_spa(episode_logs):
    successes = 0
    for episode in episode_logs:
        if episode["distance"] > 1000 and episode["collisions"] == 0:
            successes += 1
    return successes / len(episode_logs)