CARLA仿真环境下基于TD3的深度强化学习实战指南

1次阅读

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

CARLA 作为开源自动驾驶仿真平台，提供了高度可配置的传感器模拟和逼真的交通场景，但其复杂环境也带来三大挑战：

非稳态环境：动态交通参与者（车辆、行人）导致状态转移概率持续变化
高维状态空间：多传感器融合（如相机 +LiDAR）使观测维度爆炸性增长
稀疏奖励：安全驾驶的负反馈（碰撞）远多于正反馈（平稳行驶）

传统强化学习算法如 DQN 难以处理连续动作空间，而 DDPG 又面临过估计问题。这正是 TD3（Twin Delayed DDPG）的用武之地。

算法	优势	劣势	适用场景
DDPG	直接处理连续动作	Q 值过估计严重	低维状态空间
PPO	策略更新稳定	对超参数敏感	离散 / 连续动作
TD3	双 Critic 抑制过估计	实现复杂度较高	高维连续控制

TD3 通过三大改进解决 DDPG 缺陷：

双 Critic 网络：取两个 Q 值的最小值作为更新目标
策略延迟更新：Critic 收敛后再更新 Actor
目标策略平滑：对动作添加噪声防止策略坍缩

class CarlaEnv():
    def __init__(self):
        self.client = carla.Client('localhost', 2000)
        self.world = self.client.load_world('Town05')

    def get_observation(self):
        # 标准化传感器数据 (RGB:0-1, LiDAR: -1~1)
        obs = {
            'camera': camera_data / 255.0,
            'lidar': (lidar_data - MEAN) / STD
        }
        return obs

    def step(self, action):
        # 将 [-1,1] 的 action 映射到实际控制量
        steer = action[0] * MAX_STEER
        throttle = (action[1] + 1) / 2 * MAX_THROTTLE
        brake = (action[2] + 1) / 2 * MAX_BRAKE

        vehicle.apply_control(carla.VehicleControl(steer=steer, throttle=throttle, brake=brake))

        # 计算多目标奖励
        reward = 0.1 * speed + 10 * lane_keep - 100 * collision
        done = collision or timeout
        return obs, reward, done, {}

class Critic(nn.Module):
    def __init__(self, state_dim, action_dim):
        super().__init__()
        self.fc1 = nn.Linear(state_dim + action_dim, 256)
        self.fc2 = nn.Linear(256, 256)
        self.q = nn.Linear(256, 1)

    def forward(self, state, action):
        x = torch.cat([state, action], dim=1)
        x = F.relu(self.fc1(x))
        x = F.relu(self.fc2(x))
        return self.q(x)

class Actor(nn.Module):
    def __init__(self, state_dim, action_dim):
        super().__init__()
        self.fc1 = nn.Linear(state_dim, 256)
        self.fc2 = nn.Linear(256, 256)
        self.mu = nn.Linear(256, action_dim)

    def forward(self, state):
        x = F.relu(self.fc1(state))
        x = F.relu(self.fc2(x))
        return torch.tanh(self.mu(x))

class PERBuffer():
    def __init__(self, capacity, alpha=0.6):
        self.capacity = capacity
        self.alpha = alpha
        self.priorities = np.zeros(capacity)
        self.buffer = []

    def add(self, transition, error):
        priority = (error + 1e-5) ** self.alpha
        if len(self.buffer) < self.capacity:
            self.buffer.append(transition)
        else:
            idx = np.argmin(self.priorities)
            self.buffer[idx] = transition
        self.priorities[idx] = priority

    def sample(self, batch_size, beta=0.4):
        probs = self.priorities / self.priorities.sum()
        indices = np.random.choice(len(self.buffer), batch_size, p=probs)
        weights = (len(self.buffer) * probs[indices]) ** (-beta)
        weights /= weights.max()
        return [self.buffer[i] for i in indices], indices, weights

def train(self, batch, weights):
    states, actions, rewards, next_states, dones = batch

    # 计算目标 Q 值
    with torch.no_grad():
        noise = (torch.randn_like(actions) * 0.2).clamp(-0.5, 0.5)
        next_actions = (self.actor_target(next_states) + noise).clamp(-1, 1)
        q1, q2 = self.critic_target(next_states, next_actions)
        target_q = rewards + (1 - dones) * self.gamma * torch.min(q1, q2)

    # 更新 Critic
    current_q1 = self.critic1(states, actions)
    current_q2 = self.critic2(states, actions)
    critic_loss = (F.mse_loss(current_q1, target_q) + 
                   F.mse_loss(current_q2, target_q)) * weights

    # 延迟更新 Actor
    if self.total_steps % self.policy_freq == 0:
        actor_loss = -self.critic1(states, self.actor(states)).mean()
        self.actor_optimizer.zero_grad()
        actor_loss.backward()
        self.actor_optimizer.step()

        # 软更新目标网络
        self.soft_update(self.critic1, self.critic1_target)
        self.soft_update(self.actor, self.actor_target)