保姆级教程:手把手教你用Python为AWS DeepRacer 2018赛道写一个“跟线”奖励函数
2026/6/7 18:26:30 网站建设 项目流程

从零构建AWS DeepRacer跟线奖励函数的完整指南

当第一次打开AWS DeepRacer的奖励函数编辑器时,面对空白的代码区域,很多初学者会感到无从下手。本文将彻底拆解一个完整的跟线奖励函数,用工程化的思维带你理解每个代码模块的设计原理。不同于简单的代码复制粘贴,我们会从赛车物理特性、强化学习反馈机制等维度,构建一个可解释、可调试的奖励系统。

1. 环境准备与基础概念

在开始编写代码前,我们需要明确几个核心概念。DeepRacer的奖励函数本质上是一个实时评分系统,它会在车辆行驶的每一步(约15次/秒)对当前状态进行评估,并返回一个奖励值。这个值就像教练的即时反馈,告诉模型当前操作的好坏。

关键参数解析

params = { 'x': 车辆当前x坐标, 'y': 车辆当前y坐标, 'heading': 车辆朝向角度(0-360度), 'speed': 当前速度(米/秒), 'steering_angle': 方向盘角度(-30到30度), 'track_width': 赛道宽度, 'closest_waypoints': 最近的两个官方路径点索引, 'progress': 已完成赛道百分比 }

这些原始参数就像赛车的传感器数据,我们需要通过计算将其转化为有意义的评估指标。一个典型的跟线奖励函数会关注三个核心维度:

  1. 轨迹跟随:车辆是否沿着理想路线行驶
  2. 速度控制:是否在合适的位置加速/减速
  3. 行驶效率:是否用最短时间完成赛道

2. 几何计算模块实现

跟线算法的核心是计算车辆到理想赛车线的距离。这里我们采用线性代数方法,通过向量投影实现高效计算。

2.1 两点距离计算

基础距离公式是所有几何运算的基石:

def dist_2_points(x1, y1, x2, y2): """计算二维平面两点间欧氏距离""" return ((x1-x2)**2 + (y1-y2)**2)**0.5

2.2 最近路径点查找

我们需要找到赛车线上离车辆最近的两个点,这对后续的线性投影至关重要:

def closest_2_racing_points(racing_line, car_pos): """返回赛车线上距离车辆最近的两个点索引""" distances = [ dist_2_points(car_pos[0], car_pos[1], p[0], p[1]) for p in racing_line ] closest_idx = distances.index(min(distances)) # 排除最近点后找第二近点 distances[closest_idx] = float('inf') second_idx = distances.index(min(distances)) return closest_idx, second_idx

2.3 点到线段的垂直距离

这是跟线算法的核心数学原理,我们使用向量叉积公式计算:

def dist_to_line(A, B, C): """计算点C到线段AB的垂直距离""" # 向量AB AB = (B[0]-A[0], B[1]-A[1]) # 向量AC AC = (C[0]-A[0], C[1]-A[1]) # 叉积模长 |AB × AC| cross = abs(AB[0]*AC[1] - AB[1]*AC[0]) # AB长度 AB_length = dist_2_points(A[0],A[1], B[0],B[1]) return cross / AB_length if AB_length > 0 else float('inf')

这个几何模块构成了奖励函数的基础设施,后续所有评估指标都建立在这些精确计算之上。

3. 动态评估指标设计

有了基础几何计算能力后,我们需要设计具体的评估指标,将原始数据转化为有意义的奖励信号。

3.1 轨迹偏离惩罚

理想的跟线行为应该让车辆尽可能贴近预设的最优路线:

def calc_track_reward(racing_line, car_pos, track_width): closest_idx, second_idx = closest_2_racing_points(racing_line, car_pos) A = racing_line[closest_idx][:2] # 最近点坐标 B = racing_line[second_idx][:2] # 次近点坐标 distance = dist_to_line(A, B, car_pos) # 标准化到0-1范围,距离超过半赛道宽度则得最低分 reward = max(0, 1 - (distance / (track_width*0.5))) return reward

3.2 速度适配奖励

不同赛道段需要不同的速度策略,我们比较当前速度与预设最优速度的匹配程度:

def calc_speed_reward(racing_line, car_pos, current_speed): closest_idx, _ = closest_2_racing_points(racing_line, car_pos) optimal_speed = racing_line[closest_idx][2] # 预设最优速度 speed_diff = abs(optimal_speed - current_speed) # 小偏差不惩罚,大偏差二次方惩罚 if speed_diff <= 1.0: # 1m/s容忍区间 return (1 - (speed_diff/1.0)**2)**2 return 0

3.3 方向一致性检查

防止车辆出现180度调头等异常行为:

def check_direction(racing_line, car_pos, heading): closest_idx, second_idx = closest_2_racing_points(racing_line, car_pos) A = racing_line[closest_idx][:2] B = racing_line[second_idx][:2] # 计算赛道方向角度 track_angle = math.degrees(math.atan2(B[1]-A[1], B[0]-A[0])) # 计算与车辆朝向的夹角差 angle_diff = abs(track_angle - heading) angle_diff = min(angle_diff, 360-angle_diff) return angle_diff < 30 # 超过30度则认为方向错误

4. 完整奖励函数集成

将各个模块有机组合,并添加进度奖励等全局优化目标:

class RacingReward: def __init__(self, racing_line): self.racing_line = racing_line self.best_lap_time = 27.0 # 预设最佳圈速 self.base_lap_time = 37.0 # 基准圈速 def reward_function(self, params): # 基础安全校验 if not params['all_wheels_on_track']: return 1e-3 # 极小奖励 # 初始化奖励值 reward = 1.0 # 轨迹跟随奖励(权重50%) track_reward = calc_track_reward( self.racing_line, [params['x'], params['y']], params['track_width'] ) reward += track_reward * 0.5 # 速度适配奖励(权重30%) speed_reward = calc_speed_reward( self.racing_line, [params['x'], params['y']], params['speed'] ) reward += speed_reward * 0.3 # 方向一致性检查 if not check_direction( self.racing_line, [params['x'], params['y']], params['heading'] ): return 1e-3 # 进度奖励(权重20%) if params['progress'] == 100: time_ratio = (self.base_lap_time - self.best_lap_time) / self.base_lap_time reward += time_ratio * 2.0 return float(reward)

5. 高级调优技巧

基础版本实现后,可以通过以下策略进一步提升模型表现:

5.1 动态权重调整

根据赛道特性自动调整各维度权重:

def dynamic_weights(racing_line, car_pos): closest_idx, _ = closest_2_racing_points(racing_line, car_pos) segment_type = classify_segment(closest_idx) # 自定义赛道段分类函数 weights = { 'track': 0.5, 'speed': 0.3, 'progress': 0.2 } if segment_type == 'straight': weights.update({'speed': 0.5, 'track': 0.3}) elif segment_type == 'sharp_turn': weights.update({'track': 0.7, 'speed': 0.1}) return weights

5.2 平滑奖励过渡

使用指数移动平均避免奖励突变:

class SmoothReward: def __init__(self, alpha=0.3): self.alpha = alpha self.smoothed = None def update(self, new_reward): if self.smoothed is None: self.smoothed = new_reward else: self.smoothed = self.alpha*new_reward + (1-self.alpha)*self.smoothed return self.smoothed

5.3 赛道特定优化

针对不同赛道修改赛车线数据:

# 2018冠军赛道参考线 racing_line_2018 = [ [3.07857, 0.7234, 3.2, 0.04483], [3.22295, 0.71246, 3.2, 0.04525], # ...完整数据约200个点 ] # 自定义赛道适配 def adapt_racing_line(original_line, new_track): # 实现坐标系变换和速度曲线调整 pass

在实际比赛中,我们会录制人类高手的驾驶轨迹作为初始赛车线,然后通过遗传算法进一步优化每个点的理想速度。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询