共计 5130 个字符，预计需要花费 13 分钟才能阅读完成。
背景痛点
自动驾驶仿真训练中存在几个典型挑战，对于新手来说尤其需要注意：
- 动作延迟：在强化学习训练中，从感知到决策再到执行存在时间差，可能导致车辆失控。
- 传感器噪声：模拟环境中的传感器数据并不完美，需要考虑噪声的影响。
- 奖励稀疏：自动驾驶任务中，好的驾驶行为往往很难直接量化，导致奖励信号稀疏。
这些挑战使得自动驾驶强化学习的入门门槛较高，需要特别注意环境配置和算法设计。
技术选型
市面上有几个主流的自动驾驶仿真平台，比如 CARLA 和 AirSim。我们选择 CARLA 的原因主要有：
- 开源免费：CARLA 完全开源，适合个人开发者和小团队使用。
- 强化学习友好：CARLA 提供了 PythonAPI，方便与强化学习框架集成。
- 场景丰富：CARLA 内置多种城市环境和天气条件，适合训练多样化的驾驶策略。
- 社区活跃：CARLA 有较大的开发者社区，遇到问题时更容易找到解决方案。
相比之下，AirSim 虽然也是一个不错的选择，但在自动驾驶场景的丰富性和社区支持方面略逊于 CARLA。
核心实现
1. 建立强化学习环境类
首先，我们需要使用 CARLA 的 PythonAPI 建立一个强化学习环境类。这个类负责与 CARLA 服务器通信，处理观测数据，并执行动作。
import carla
import numpy as np
from typing import Tuple, Dict, Any
class CarlaEnv:
"""CARLA 强化学习环境类"""
def __init__(self, host: str = '127.0.0.1', port: int = 2000):
"""初始化 CARLA 环境"""
self.client = carla.Client(host, port)
self.client.set_timeout(10.0)
self.world = self.client.get_world()
self.blueprint_library = self.world.get_blueprint_library()
def reset(self) -> np.ndarray:
"""重置环境，返回初始观测"""
# 清除现有车辆和传感器
self._cleanup()
# 生成车辆
vehicle_bp = self.blueprint_library.find('vehicle.tesla.model3')
spawn_point = self.world.get_map().get_spawn_points()[0]
self.vehicle = self.world.spawn_actor(vehicle_bp, spawn_point)
# 添加 RGB 相机
camera_bp = self.blueprint_library.find('sensor.camera.rgb')
camera_bp.set_attribute('image_size_x', '800')
camera_bp.set_attribute('image_size_y', '600')
camera_transform = carla.Transform(carla.Location(x=1.5, z=2.4))
self.camera = self.world.spawn_actor(camera_bp, camera_transform, attach_to=self.vehicle)
# 设置观测缓冲区
self.image_queue = queue.Queue()
self.camera.listen(self.image_queue.put)
# 获取初始观测
return self._get_observation()
def _get_observation(self) -> np.ndarray:
"""获取当前观测（RGB 图像）"""
image = self.image_queue.get()
array = np.frombuffer(image.raw_data, dtype=np.uint8)
array = np.reshape(array, (image.height, image.width, 4))
return array[:, :, :3] # 去掉 alpha 通道2. 实现 DQN 网络
接下来，我们基于 PyTorch 实现 DQN 网络，包含经验回放机制。
import torch
import torch.nn as nn
import torch.optim as optim
import random
from collections import deque
from typing import Deque, Tuple
class DQN(nn.Module):
"""深度 Q 网络"""
def __init__(self, input_shape: Tuple[int, int, int], n_actions: int):
super(DQN, self).__init__()
self.conv = nn.Sequential(nn.Conv2d(input_shape[2], 32, kernel_size=8, stride=4),
nn.ReLU(),
nn.Conv2d(32, 64, kernel_size=4, stride=2),
nn.ReLU(),
nn.Conv2d(64, 64, kernel_size=3, stride=1),
nn.ReLU())
conv_out_size = self._get_conv_out(input_shape)
self.fc = nn.Sequential(nn.Linear(conv_out_size, 512),
nn.ReLU(),
nn.Linear(512, n_actions)
)
def _get_conv_out(self, shape: Tuple[int, int, int]) -> int:
"""计算卷积层输出大小"""
o = self.conv(torch.zeros(1, *shape))
return int(np.prod(o.size()))
def forward(self, x: torch.Tensor) -> torch.Tensor:
"""前向传播"""
conv_out = self.conv(x.permute(0, 3, 1, 2).float())
return self.fc(conv_out.view(conv_out.size(0), -1))
class ReplayBuffer:
"""经验回放缓冲区"""
def __init__(self, capacity: int = 10000):
self.buffer: Deque[Tuple] = deque(maxlen=capacity)
def push(self, transition: Tuple) -> None:
"""存储转换"""
self.buffer.append(transition)
def sample(self, batch_size: int) -> Tuple:
"""随机采样一批转换"""
batch = random.sample(self.buffer, batch_size)
state, action, reward, next_state, done = zip(*batch)
return torch.stack(state), torch.tensor(action), torch.tensor(reward), torch.stack(next_state), torch.tensor(done)
def __len__(self) -> int:
return len(self.buffer)3. 训练循环
设置关键超参数 γ =0.99, batch_size=64，实现训练循环：
def train(env: CarlaEnv, dqn: DQN, target_dqn: DQN, buffer: ReplayBuffer,
optimizer: optim.Optimizer, gamma: float = 0.99, batch_size: int = 64):
"""训练循环"""
epsilon = 1.0
epsilon_min = 0.01
epsilon_decay = 0.995
for episode in range(1000):
state = env.reset()
episode_reward = 0
while True:
# ε- 贪婪策略选择动作
if random.random() < epsilon:
action = env.action_space.sample()
else:
with torch.no_grad():
q_values = dqn(torch.tensor(state).unsqueeze(0))
action = q_values.argmax().item()
# 执行动作，获取下一个状态和奖励
next_state, reward, done, _ = env.step(action)
episode_reward += reward
# 存储转换
buffer.push((torch.tensor(state), action, reward, torch.tensor(next_state), done))
# 如果缓冲区足够大，开始训练
if len(buffer) >= batch_size:
states, actions, rewards, next_states, dones = buffer.sample(batch_size)
# 计算当前 Q 值和目标 Q 值
current_q = dqn(states).gather(1, actions.unsqueeze(1))
next_q = target_dqn(next_states).max(1)[0].detach()
target_q = rewards + gamma * next_q * (1 - dones)
# 计算损失并更新网络
loss = nn.MSELoss()(current_q.squeeze(), target_q)
optimizer.zero_grad()
loss.backward()
optimizer.step()
state = next_state
if done:
break
# 更新目标网络
if episode % 10 == 0:
target_dqn.load_state_dict(dqn.state_dict())
# 衰减 ε
epsilon = max(epsilon_min, epsilon * epsilon_decay)
print(f"Episode {episode}, Reward: {episode_reward}, Epsilon: {epsilon:.2f}")避坑指南
在 CARLA 强化学习实践中，有几个常见问题需要特别注意：
- 时钟同步问题：
- CARLA 服务器和客户端可能存在时钟不同步的情况。
-
解决方案：使用
client.set_timeout()设置合理的超时时间，并在关键操作处添加重试机制。 -
动作空间离散化导致的 ” 抖动驾驶 ”：
- 当将连续动作空间离散化时，可能导致车辆控制不流畅。
-
解决方案：增加离散动作的数量，或考虑使用 DDPG 等连续动作空间的算法。
-
多传感器数据对齐：
- 不同传感器的数据可能不是同时生成的。
- 解决方案：使用 CARLA 的同步模式(
world.tick())，或基于时间戳进行数据对齐。
性能优化
- 使用 Ray 进行分布式采样：
- 可以显著加快数据收集速度。
-
示例代码：
import ray @ray.remote class CarlaWorker: def __init__(self): self.env = CarlaEnv() def collect_data(self, policy, n_steps): # 实现数据收集逻辑 pass -
观测数据归一化：
- 对图像数据进行归一化 (如除以 255) 可以提高训练稳定性。
- 示例：
image = image / 255.0
验证指标
训练过程中需要监控几个关键指标：
- 回合奖励：随着训练应该逐渐增加。
- 成功率：测试阶段完成任务的比率。
- Q 值变化：可以帮助判断算法是否收敛。
可以绘制这些指标的变化曲线来评估训练效果。
延伸思考
仿真环境中训练的策略如何迁移到真实车辆？这是一个典型的域适配问题，可以考虑以下方法：
- 域随机化：在仿真中引入更多样的场景和传感器噪声。
- 模型微调：先在仿真中预训练，然后在真实数据上微调。
- 中间表示：使用不依赖于具体传感器的中间表示(如语义分割图)。
总结
本文详细介绍了如何在 CARLA 仿真平台上实现自动驾驶强化学习。从环境搭建到 DQN 算法实现，再到性能优化和验证，涵盖了整个流程的关键环节。希望这篇指南能帮助初学者顺利入门 CARLA 强化学习，为后续更复杂的研究打下基础。
在实际应用中，建议从简单场景开始，逐步增加复杂度。同时，多尝试不同的算法和参数配置，找到最适合特定任务的方法。
正文完
