diff --git a/src/Unmanned_vehicle_AD_DQN/Hyperparameters.py b/src/Unmanned_vehicle_AD_DQN/Hyperparameters.py index d7602f3f4e..bf5be3a954 100644 --- a/src/Unmanned_vehicle_AD_DQN/Hyperparameters.py +++ b/src/Unmanned_vehicle_AD_DQN/Hyperparameters.py @@ -86,9 +86,25 @@ PER_BETA_FRAMES = 100000 # beta线性增长的帧数 -# Dueling DQN 参数 -USE_DUELING = True -# 是否使用Dueling DQN架构 +# 训练策略参数 +USE_CURRICULUM_LEARNING = True +# 是否使用课程学习 -USE_PER = True -# 是否使用优先经验回放 \ No newline at end of file +USE_MULTI_OBJECTIVE = True +# 是否使用多目标优化 + +USE_IMITATION_LEARNING = False +# 是否使用模仿学习（预训练） + +# 多目标优化权重（这些权重会自动调整） +SAFETY_WEIGHT = 0.40 +EFFICIENCY_WEIGHT = 0.25 +COMFORT_WEIGHT = 0.20 +RULE_FOLLOWING_WEIGHT = 0.15 + +# 课程学习参数 +CURRICULUM_STAGES = 5 +# 课程学习阶段数量 + +CURRICULUM_SUCCESS_THRESHOLDS = [0.3, 0.5, 0.7, 0.85, 0.9] +# 每个阶段进入下一阶段所需的成功率阈值 \ No newline at end of file diff --git a/src/Unmanned_vehicle_AD_DQN/Model.py b/src/Unmanned_vehicle_AD_DQN/Model.py index 3f8f1675b6..c419ae6c35 100644 --- a/src/Unmanned_vehicle_AD_DQN/Model.py +++ b/src/Unmanned_vehicle_AD_DQN/Model.py @@ -19,6 +19,9 @@ from threading import Thread from Environment import * from Hyperparameters import * +import pickle +import json +from datetime import datetime # 自定义TensorBoard类 @@ -116,12 +119,445 @@ def update_priorities(self, indices, errors): self.priorities[idx] = (abs(error) + 1e-5) ** self.alpha -# DQN智能体类 - 升级版 +# 课程学习管理器 +class CurriculumManager: + def __init__(self, env): + self.env = env + self.current_stage = 0 + self.stage_thresholds = [0.3, 0.5, 0.7, 0.85] # 成功率阈值 + self.stage_configs = [ + # 阶段0: 入门 + { + 'pedestrian_cross': 4, # 十字路口行人数量 + 'pedestrian_normal': 2, # 普通路段行人数量 + 'pedestrian_speed_min': 0.5, # 行人最低速度 + 'pedestrian_speed_max': 1.0, # 行人最高速度 + 'max_episode_steps': 1200, # 最大步数 (20秒 * 60FPS) + 'success_threshold': 0.3 # 进入下一阶段成功率 + }, + # 阶段1: 初级 + { + 'pedestrian_cross': 6, + 'pedestrian_normal': 3, + 'pedestrian_speed_min': 0.7, + 'pedestrian_speed_max': 1.3, + 'max_episode_steps': 1800, # 30秒 + 'success_threshold': 0.5 + }, + # 阶段2: 中级 + { + 'pedestrian_cross': 8, + 'pedestrian_normal': 4, + 'pedestrian_speed_min': 0.8, + 'pedestrian_speed_max': 1.5, + 'max_episode_steps': 2400, # 40秒 + 'success_threshold': 0.7 + }, + # 阶段3: 高级 (正常难度) + { + 'pedestrian_cross': 10, + 'pedestrian_normal': 5, + 'pedestrian_speed_min': 1.0, + 'pedestrian_speed_max': 2.0, + 'max_episode_steps': 3600, # 60秒 + 'success_threshold': 0.85 + }, + # 阶段4: 专家 (挑战) + { + 'pedestrian_cross': 12, + 'pedestrian_normal': 6, + 'pedestrian_speed_min': 1.2, + 'pedestrian_speed_max': 2.5, + 'max_episode_steps': 3600, + 'success_threshold': 0.9 + } + ] + + # 训练历史 + self.success_history = deque(maxlen=20) # 记录最近20轮的成功情况 + self.reward_history = deque(maxlen=50) # 记录最近50轮的奖励 + + def update_stage(self, success, reward): + """更新训练阶段""" + # 记录历史 + self.success_history.append(1 if success else 0) + self.reward_history.append(reward) + + # 计算最近成功率 + if len(self.success_history) >= 10: + success_rate = sum(self.success_history) / len(self.success_history) + avg_reward = np.mean(self.reward_history) if self.reward_history else 0 + + print(f"课程学习 - 当前阶段: {self.current_stage}, 成功率: {success_rate:.2f}, 平均奖励: {avg_reward:.2f}") + + # 检查是否可以进入下一阶段 + if self.current_stage < len(self.stage_configs) - 1: + next_stage_threshold = self.stage_configs[self.current_stage]['success_threshold'] + if success_rate >= next_stage_threshold and avg_reward > 5: + self.current_stage += 1 + print(f"🎉 课程学习: 进阶到阶段 {self.current_stage}!") + return True + + # 如果表现太差，退回上一阶段 + if self.current_stage > 0 and success_rate < 0.2: + self.current_stage -= 1 + print(f"⚠️ 课程学习: 退回阶段 {self.current_stage}") + return True + + return False + + def get_current_config(self): + """获取当前阶段的配置""" + return self.stage_configs[min(self.current_stage, len(self.stage_configs) - 1)] + + def apply_to_environment(self): + """将当前阶段配置应用到环境""" + config = self.get_current_config() + # 注意：这里需要修改Environment.py中的行人生成逻辑来支持这些参数 + # 暂时返回配置，由外部调用者处理 + return config + + +# 多目标优化器 +class MultiObjectiveOptimizer: + def __init__(self): + # 定义优化目标及其权重（可动态调整） + self.objectives = { + 'safety': { + 'weight': 0.4, + 'description': '安全避障和避免碰撞', + 'metrics': ['collision_avoidance', 'pedestrian_distance'] + }, + 'efficiency': { + 'weight': 0.25, + 'description': '快速到达目的地', + 'metrics': ['progress_speed', 'total_time'] + }, + 'comfort': { + 'weight': 0.2, + 'description': '平稳驾驶体验', + 'metrics': ['smoothness', 'steering_changes'] + }, + 'rule_following': { + 'weight': 0.15, + 'description': '遵守交通规则', + 'metrics': ['lane_keeping', 'speed_limit'] + } + } + + # 指标跟踪 + self.metrics_history = { + 'safety': [], + 'efficiency': [], + 'comfort': [], + 'rule_following': [] + } + + def compute_composite_reward(self, metrics): + """计算综合奖励值""" + composite = 0 + + for obj_name, obj_info in self.objectives.items(): + if obj_name in metrics: + # 归一化处理每个目标的贡献 + normalized_value = self._normalize_metric(metrics[obj_name], obj_name) + composite += normalized_value * obj_info['weight'] + + # 记录指标历史 + self.metrics_history[obj_name].append(normalized_value) + + # 特殊惩罚项 + if metrics.get('collision', False): + composite -= 10 + if metrics.get('off_road', False): + composite -= 5 + if metrics.get('dangerous_action', False): + composite -= 3 + + return composite + + def _normalize_metric(self, value, metric_name): + """归一化指标值到[0, 1]范围""" + # 不同指标的归一化方式不同 + normalization_rules = { + 'safety': lambda x: min(max(x / 10, 0), 1), # 假设安全分满分10 + 'efficiency': lambda x: min(max(x / 100, 0), 1), # 效率分满分100 + 'comfort': lambda x: min(max((x + 5) / 10, 0), 1), # 舒适度[-5, 5] -> [0, 1] + 'rule_following': lambda x: min(max(x, 0), 1) # 规则遵循度[0, 1] + } + + if metric_name in normalization_rules: + return normalization_rules[metric_name](value) + return min(max(value, 0), 1) # 默认截断到[0, 1] + + def adjust_weights(self, performance_feedback): + """根据性能反馈动态调整权重""" + # 如果某个目标表现持续较差，增加其权重 + recent_performance = {} + for obj in self.objectives: + if len(self.metrics_history[obj]) >= 10: + recent_avg = np.mean(self.metrics_history[obj][-10:]) + recent_performance[obj] = recent_avg + + if recent_performance: + # 找到表现最差的目标 + worst_obj = min(recent_performance, key=recent_performance.get) + best_obj = max(recent_performance, key=recent_performance.get) + + # 如果最差目标表现低于阈值，增加其权重 + if recent_performance[worst_obj] < 0.3: + adjustment = 0.05 + self.objectives[worst_obj]['weight'] += adjustment + self.objectives[best_obj]['weight'] -= adjustment + + # 确保权重总和为1 + total = sum(obj['weight'] for obj in self.objectives.values()) + for obj in self.objectives: + self.objectives[obj]['weight'] /= total + + print(f"动态权重调整: {worst_obj}权重↑ {adjustment:.3f}, {best_obj}权重↓ {adjustment:.3f}") + + def get_performance_report(self): + """生成性能报告""" + report = "多目标优化性能报告:\n" + report += "=" * 50 + "\n" + + for obj_name, obj_info in self.objectives.items(): + history = self.metrics_history[obj_name] + if history: + avg = np.mean(history[-20:]) if len(history) >= 20 else np.mean(history) + report += f"{obj_name}(权重:{obj_info['weight']:.2f}): 平均得分={avg:.3f}\n" + report += f" 描述: {obj_info['description']}\n" + + return report + + +# 模仿学习管理器 +class ImitationLearningManager: + def __init__(self, expert_data_path=None): + self.expert_data_path = expert_data_path + self.expert_data = [] + self.is_pretrained = False + + def load_expert_data(self, path): + """加载专家示范数据""" + try: + if os.path.exists(path): + with open(path, 'rb') as f: + self.expert_data = pickle.load(f) + print(f"已加载 {len(self.expert_data)} 条专家示范数据") + return True + else: + print(f"专家数据文件不存在: {path}") + return False + except Exception as e: + print(f"加载专家数据失败: {e}") + return False + + def collect_expert_demonstration(self, env, num_episodes=10): + """收集专家示范数据（可以手动控制或使用规则控制器）""" + print(f"开始收集专家示范数据 ({num_episodes}个episodes)...") + + demonstrations = [] + + for episode in range(num_episodes): + print(f"收集专家示范 Episode {episode + 1}/{num_episodes}") + + state = env.reset(episode) + done = False + episode_data = [] + + while not done: + # 这里可以使用规则控制器或手动控制 + # 示例：简单的规则控制器 + action = self._rule_based_controller(env) + + new_state, reward, done, _ = env.step(action) + + # 保存示范数据 + episode_data.append({ + 'state': state.copy(), + 'action': action, + 'reward': reward, + 'next_state': new_state.copy(), + 'done': done + }) + + state = new_state + + demonstrations.extend(episode_data) + env.cleanup_actors() + + # 保存专家数据 + self.expert_data = demonstrations + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + save_path = f"expert_data_{timestamp}.pkl" + + with open(save_path, 'wb') as f: + pickle.dump(demonstrations, f) + + print(f"专家示范数据已保存到: {save_path}, 共 {len(demonstrations)} 条记录") + return True + + def _rule_based_controller(self, env): + """基于规则的控制器（作为专家示范）""" + # 获取车辆状态 + vehicle_location = env.vehicle.get_location() + velocity = env.vehicle.get_velocity() + speed_kmh = 3.6 * math.sqrt(velocity.x**2 + velocity.y**2) + + # 简单规则：保持速度在20-40 km/h，避免障碍物 + if speed_kmh < 20: + return 2 # 加速 + elif speed_kmh > 40: + return 0 # 减速 + else: + # 检查前方障碍物 + has_obstacle_ahead = self._check_obstacle_ahead(env) + if has_obstacle_ahead: + return 0 # 减速 + else: + return 1 # 保持 + + return 1 # 默认保持 + + def _check_obstacle_ahead(self, env): + """检查前方是否有障碍物（简化版本）""" + # 这里可以添加更复杂的障碍物检测逻辑 + # 暂时返回False + return False + + def pretrain_with_behavioral_cloning(self, model, epochs=20): + """使用行为克隆进行预训练""" + if not self.expert_data: + print("没有专家数据可用，跳过预训练") + return model + + print(f"开始行为克隆预训练 ({epochs}个epochs)...") + + # 准备训练数据 + states = [] + actions = [] + + for demo in self.expert_data: + states.append(demo['state']) + actions.append(demo['action']) + + # 将状态归一化 + states = np.array(states) / 255.0 + + # 将动作转换为one-hot编码 + actions_onehot = tf.keras.utils.to_categorical(actions, num_classes=5) + + # 备份原始编译设置 + original_loss = model.loss + original_optimizer = model.optimizer + original_metrics = model.metrics_names + + # 重新编译模型用于分类任务 + model.compile( + optimizer=Adam(learning_rate=0.0001), + loss='categorical_crossentropy', + metrics=['accuracy'] + ) + + # 训练模型模仿专家行为 + history = model.fit( + states, actions_onehot, + batch_size=32, + epochs=epochs, + validation_split=0.2, + verbose=1 + ) + + print(f"预训练完成 - 最终准确率: {history.history['accuracy'][-1]:.3f}") + + # 恢复原始编译设置 + model.compile( + optimizer=original_optimizer, + loss=original_loss, + metrics=original_metrics + ) + + self.is_pretrained = True + return model + + def train_with_dagger(self, model, env, iterations=5, episodes_per_iter=5): + """使用DAgger算法进行训练""" + print(f"开始DAgger训练 ({iterations}次迭代，每次{episodes_per_iter}个episodes)...") + + aggregated_data = self.expert_data.copy() + + for iteration in range(iterations): + print(f"\nDAgger 迭代 {iteration + 1}/{iterations}") + + # 使用当前策略收集数据 + new_demos = [] + + for episode in range(episodes_per_iter): + print(f" 收集数据 Episode {episode + 1}/{episodes_per_iter}") + + state = env.reset(episode) + done = False + + while not done: + # 使用当前策略选择动作 + qs = model.predict(np.array(state).reshape(-1, *state.shape) / 255)[0] + action = np.argmax(qs) + + # 执行动作 + new_state, reward, done, _ = env.step(action) + + # 专家纠正（这里可以添加专家纠正逻辑） + # 如果策略动作与专家建议不同，使用专家动作 + expert_action = self._rule_based_controller(env) + + # 保存数据（使用专家纠正后的动作） + new_demos.append({ + 'state': state.copy(), + 'action': expert_action, # 使用专家动作 + 'reward': reward, + 'next_state': new_state.copy(), + 'done': done + }) + + state = new_state + + env.cleanup_actors() + + # 合并数据 + aggregated_data.extend(new_demos) + + # 在合并数据上重新训练 + states = [d['state'] for d in aggregated_data] + actions = [d['action'] for d in aggregated_data] + + states = np.array(states) / 255.0 + actions_onehot = tf.keras.utils.to_categorical(actions, num_classes=5) + + # 训练模型 + history = model.fit( + states, actions_onehot, + batch_size=32, + epochs=10, + validation_split=0.1, + verbose=0 + ) + + print(f" 训练完成 - 准确率: {history.history['accuracy'][-1]:.3f}") + + print("DAgger训练完成!") + return model + + +# DQN智能体类 - 升级版（整合训练策略） class DQNAgent: - def __init__(self, use_dueling=True, use_per=True): + def __init__(self, use_dueling=True, use_per=True, use_curriculum=True, use_multi_objective=True): # 创建主网络和目标网络 self.use_dueling = use_dueling self.use_per = use_per + self.use_curriculum = use_curriculum + self.use_multi_objective = use_multi_objective if use_dueling: self.model = self.create_dueling_model() @@ -141,12 +577,29 @@ def __init__(self, use_dueling=True, use_per=True): # 自定义TensorBoard self.tensorboard = ModifiedTensorBoard(log_dir=f"logs/{MODEL_NAME}-{int(time.time())}") self.target_update_counter = 0 # 目标网络更新计数器 - self.graph = tf.compat.v1.get_default_graph() # 训练控制标志 self.terminate = False self.last_logged_episode = 0 self.training_initialized = False + + # 训练策略组件 + self.curriculum_manager = None + self.multi_objective_optimizer = None + self.imitation_manager = None + + def setup_training_strategies(self, env=None): + """设置训练策略组件""" + if self.use_curriculum and env: + self.curriculum_manager = CurriculumManager(env) + print("课程学习管理器已启用") + + if self.use_multi_objective: + self.multi_objective_optimizer = MultiObjectiveOptimizer() + print("多目标优化器已启用") + + # 模仿学习管理器（需要时手动启用） + self.imitation_manager = ImitationLearningManager() def create_model(self): """创建标准深度Q网络模型""" @@ -244,7 +697,6 @@ def create_dueling_model(self): advantage = Dense(5, activation='linear', name='advantage')(advantage_stream) # 合并: Q(s,a) = V(s) + (A(s,a) - mean(A(s,a))) - # 减去均值使得优势函数的均值为0 mean_advantage = Lambda(lambda a: tf.reduce_mean(a, axis=1, keepdims=True))(advantage) advantage_centered = Subtract()([advantage, mean_advantage]) q_values = Add()([value, advantage_centered]) diff --git a/src/Unmanned_vehicle_AD_DQN/Test.py b/src/Unmanned_vehicle_AD_DQN/Test.py index dec5ce76df..3c3b504f92 100644 --- a/src/Unmanned_vehicle_AD_DQN/Test.py +++ b/src/Unmanned_vehicle_AD_DQN/Test.py @@ -1,7 +1,5 @@ # Test.py import random -import os -import glob from collections import deque import numpy as np import cv2 @@ -11,413 +9,528 @@ from tensorflow.keras.models import load_model from Environment import CarEnv, MEMORY_FRACTION from Hyperparameters import * +import os +import json +import glob -def find_latest_model(model_dir=None, pattern="*.model"): +def find_model_files(model_dir="models", pattern="*.model"): """ - 自动查找最新训练的模型 - - Args: - model_dir: 模型目录路径，如果为None则使用默认目录 - pattern: 模型文件匹配模式 - - Returns: - 最新模型的路径，如果没有找到则返回None + 自动查找模型文件 """ - if model_dir is None: - model_dir = r'D:\Robots\nn\src\Unmanned_vehicle_AD_DQN\models' - - if not os.path.exists(model_dir): - print(f"警告: 模型目录不存在: {model_dir}") - return None - - # 查找所有模型文件 - model_files = glob.glob(os.path.join(model_dir, pattern)) - - if not model_files: - print(f"警告: 在目录 {model_dir} 中没有找到模型文件") - return None - - # 按修改时间排序，获取最新的模型 - latest_model = max(model_files, key=os.path.getmtime) - - # 也可以按文件名中的数字排序（如果文件名包含训练步数或episode数） - # 例如: model_1000.model, model_2000.model - try: - # 尝试按文件名中的数字排序 - def extract_number(filename): - import re - numbers = re.findall(r'\d+', os.path.basename(filename)) - return int(numbers[-1]) if numbers else 0 - - # 按数字大小排序，获取最大的（通常是最新的） - latest_by_name = max(model_files, key=extract_number) - - # 如果按名称找到的比按时间找到的更新（数字更大），则使用按名称找到的 - if extract_number(latest_by_name) > extract_number(latest_model): - latest_model = latest_by_name - print(f"按文件名排序选择模型: {os.path.basename(latest_model)}") - else: - print(f"按修改时间选择模型: {os.path.basename(latest_model)}") - except: - print(f"按修改时间选择模型: {os.path.basename(latest_model)}") + # 检查多种可能的目录结构 + possible_paths = [ + model_dir, # 当前目录下的models + os.path.join(os.path.dirname(__file__), model_dir), # 脚本同级目录 + os.path.join(os.path.dirname(__file__), "..", model_dir), # 上一级目录 + "../models", # 相对路径 + "./models", # 当前目录 + "models", # 直接models目录 + ] + + model_files = [] + + for path in possible_paths: + if os.path.exists(path): + files = glob.glob(os.path.join(path, pattern)) + if files: + print(f"在目录 '{path}' 中找到 {len(files)} 个模型文件") + model_files.extend(files) + + # 去重 + model_files = list(set(model_files)) + + # 按修改时间排序（最新的在前面） + model_files.sort(key=os.path.getmtime, reverse=True) - return latest_model + return model_files -def list_available_models(model_dir=None): +def select_best_model(model_files, preferred_keywords=None, excluded_keywords=None): """ - 列出所有可用的模型 + 从模型文件列表中选择最佳模型 + """ + if not model_files: + return None - Args: - model_dir: 模型目录路径 + if preferred_keywords is None: + preferred_keywords = ["best", "advanced", "dueling_per"] - Returns: - 模型文件列表，按修改时间排序 - """ - if model_dir is None: - model_dir = r'D:\Robots\nn\src\Unmanned_vehicle_AD_DQN\models' + if excluded_keywords is None: + excluded_keywords = ["min", "avg", "final"] # 排除统计文件 - if not os.path.exists(model_dir): - print(f"警告: 模型目录不存在: {model_dir}") - return [] + # 评分系统：根据关键词和文件属性给模型打分 + scored_models = [] - model_files = glob.glob(os.path.join(model_dir, "*.model")) + for file_path in model_files: + filename = os.path.basename(file_path) + score = 0 + + # 基于文件名关键词打分 + for keyword in preferred_keywords: + if keyword.lower() in filename.lower(): + score += 10 + + # 排除包含特定关键词的文件 + exclude = False + for keyword in excluded_keywords: + if keyword.lower() in filename.lower() and not filename.lower().endswith(".model"): + exclude = True + break + + if exclude: + continue + + # 基于文件大小和修改时间打分 + try: + file_size = os.path.getsize(file_path) / (1024 * 1024) # MB + if file_size > 100: # 大于100MB的模型可能更复杂 + score += 5 + + # 文件修改时间（越新越好） + days_old = (time.time() - os.path.getmtime(file_path)) / (24 * 3600) + if days_old < 7: # 一周内的文件 + score += 10 + elif days_old < 30: # 一个月内的文件 + score += 5 + except: + pass + + scored_models.append((file_path, score, filename)) - if not model_files: - print(f"目录 {model_dir} 中没有模型文件") - return [] + if not scored_models: + return None - # 按修改时间排序（最新的在前） - model_files.sort(key=os.path.getmtime, reverse=True) + # 按分数排序 + scored_models.sort(key=lambda x: x[1], reverse=True) - print("\n可用的模型文件:") - print("-" * 80) - for i, model_file in enumerate(model_files): - filename = os.path.basename(model_file) - mtime = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(os.path.getmtime(model_file))) - size = os.path.getsize(model_file) / (1024 * 1024) # 转换为MB - print(f"{i+1:3d}. {filename:50s} | 修改时间: {mtime} | 大小: {size:.1f} MB") - print("-" * 80) + print("\n找到的模型文件（按优先级排序）:") + for i, (path, score, name) in enumerate(scored_models[:5]): # 显示前5个 + print(f" {i+1}. [{score:3d}分] {name}") - return model_files + return scored_models[0][0] # 返回最佳模型的路径 -def get_safe_action_improved(model, state, env, previous_action, uncertainty_threshold=1.0): +def get_safe_action_advanced(model, state, env, previous_action, uncertainty_threshold=1.0): """ - 改进的安全动作选择，结合模型预测、安全规则和不确定性估计 + 高级安全动作选择，结合模型预测、安全规则、不确定性估计和多目标优化 """ # 模型预测 state_normalized = np.array(state).reshape(-1, *state.shape) / 255 qs = model.predict(state_normalized, verbose=0)[0] - # 计算不确定性（如果模型有多个输出头） - uncertainty = 0.0 - if hasattr(model, 'output'): - if isinstance(model.output, list): - # 如果是Dueling DQN，可以分别获取价值和优势 - predictions = model.predict(state_normalized, verbose=0) - if isinstance(predictions, list): - value = predictions[0][0] if len(predictions) > 0 else 0 - advantage = predictions[1][0] if len(predictions) > 1 else np.zeros(5) - # 计算不确定性作为价值和优势的差异 - uncertainty = np.std(advantage) - - # 安全规则：高不确定性时更加保守 - if uncertainty > uncertainty_threshold: - # 降低激进动作的Q值 - qs[2] *= 0.5 # 降低加速倾向 - qs[3] *= 0.7 # 降低左转倾向 - qs[4] *= 0.7 # 降低右转倾向 - - # 如果有建议的避让动作（来自环境），优先考虑 - if hasattr(env, 'suggested_action') and env.suggested_action is not None: - suggested_q = qs[env.suggested_action] - qs[env.suggested_action] += 2.0 # 大幅提高建议动作的Q值 - print(f"安全建议: 执行动作 {env.suggested_action} 以避让行人") - env.suggested_action = None # 重置 - - # 避免频繁切换动作（平滑性） - if previous_action in [3, 4]: # 如果是转向动作 - qs[previous_action] += 0.5 # 提高继续当前转向的倾向 - - # 防止过度转向 - if env.same_steer_counter > 3: # 连续同向转向超过3次 - qs[previous_action] -= 1.0 # 降低当前转向动作的Q值 - - # 速度相关的动作调整 + # 获取车辆速度 velocity = env.vehicle.get_velocity() speed_kmh = 3.6 * np.linalg.norm([velocity.x, velocity.y, velocity.z]) - if speed_kmh > 40: # 高速时更加谨慎 - qs[2] *= 0.8 # 降低加速倾向 - elif speed_kmh < 10: # 低速时鼓励加速 - qs[0] *= 0.7 # 降低减速倾向 - qs[2] *= 1.2 # 提高加速倾向 + # 1. 速度自适应调整 + speed_factor = max(0.3, min(1.0, 30.0 / max(1.0, speed_kmh))) + + if speed_kmh > 40: # 高速时更加保守 + qs[2] *= 0.6 # 降低加速倾向 + qs[3] *= 0.5 # 大幅降低左转倾向 + qs[4] *= 0.5 # 大幅降低右转倾向 + elif speed_kmh < 10: # 低速时鼓励前进 + qs[0] *= 0.5 # 降低减速倾向 + qs[1] *= 1.2 # 提高保持倾向 + qs[2] *= 1.3 # 提高加速倾向 + + # 2. 行人避障优先级 + if hasattr(env, 'suggested_action') and env.suggested_action is not None: + qs[env.suggested_action] += 3.0 # 大幅提高建议动作的Q值 + print(f"🚨 安全避让: 执行动作 {env.suggested_action}") + env.suggested_action = None + + # 3. 防止过度转向 + if hasattr(env, 'same_steer_counter') and env.same_steer_counter > 2: + if previous_action in [3, 4]: + qs[previous_action] -= 1.5 # 降低连续同向转向的倾向 + + # 4. 动作平滑性 + if previous_action in [3, 4]: # 转向动作 + qs[previous_action] += 0.8 * speed_factor # 速度相关的平滑性 + elif previous_action in [0, 2]: # 加减速动作 + qs[previous_action] += 0.3 # 轻微的惯性保持 + + # 5. 道路保持倾向 + # 如果车辆方向偏差小，鼓励保持直行 + if hasattr(env, 'vehicle'): + vehicle_rotation = env.vehicle.get_transform().rotation.yaw + if abs(vehicle_rotation) < 10: # 方向良好 + qs[1] += 0.5 # 鼓励保持 + elif abs(vehicle_rotation) > 30: # 方向偏差大 + # 鼓励向相反方向转向以回正 + if vehicle_rotation > 0: # 偏左，鼓励右转 + qs[4] += 1.0 + else: # 偏右，鼓励左转 + qs[3] += 1.0 + + # 6. 紧急情况处理 + min_ped_distance = getattr(env, 'last_ped_distance', float('inf')) + if min_ped_distance < 5.0: # 紧急避让距离 + # 大幅调整Q值以确保安全 + qs[0] += 2.0 # 紧急制动 + if min_ped_distance < 3.0: # 极危险 + qs[2] = -float('inf') # 禁止加速 + print("⚠️ 紧急制动!") # 选择动作 action = np.argmax(qs) - # 安全检查：避免危险动作 - if speed_kmh > 35 and action in [3, 4]: # 高速时避免急转 - # 检查是否有更安全的替代动作 - alternative_actions = [1, 0, 2] # 保持、减速、加速 - safe_qs = [qs[a] for a in alternative_actions] - if max(safe_qs) > qs[action] * 0.8: # 如果安全动作的Q值接近 - action = alternative_actions[np.argmax(safe_qs)] - print(f"安全调整: 高速时避免急转，选择动作 {action}") - - return action, qs, uncertainty + # 最终安全检查 + if speed_kmh > 35 and action in [3, 4]: + # 高速急转检查 + steer_magnitude = abs(qs[3]) if action == 3 else abs(qs[4]) + if steer_magnitude > 2.0: # 急转倾向强 + # 考虑更安全的替代动作 + safe_alternatives = [1, 0] # 保持或减速 + safe_qs = [qs[a] for a in safe_alternatives] + if max(safe_qs) > qs[action] * 0.7: + action = safe_alternatives[np.argmax(safe_qs)] + print(f"安全调整: 高速时避免急转，选择动作 {action}") + + return action, qs -def select_model_interactively(): - """ - 交互式选择模型 - """ - model_dir = r'D:\Robots\nn\src\Unmanned_vehicle_AD_DQN\models' - - # 列出所有可用模型 - model_files = list_available_models(model_dir) - - if not model_files: - print("没有找到模型文件，请手动指定模型路径。") - manual_path = input("请输入模型完整路径: ").strip() - if os.path.exists(manual_path): - return manual_path +def run_test_episode(model, env, episode_num, use_advanced_safety=True): + """运行单个测试episode""" + print(f"\n{'='*50}") + print(f"测试 Episode {episode_num}") + print(f"{'='*50}") + + # 重置环境 + current_state = env.reset(401) # 正常难度 + env.collision_hist = [] + + # 初始化统计 + total_reward = 0 + step_count = 0 + done = False + previous_action = 1 + fps_counter = deque(maxlen=30) + + # 运行episode + max_steps = SECONDS_PER_EPISODE * 60 + + while not done and step_count < max_steps: + step_start = time.time() + + # 选择动作 + if use_advanced_safety: + action, qs = get_safe_action_advanced(model, current_state, env, previous_action) else: - print(f"错误: 文件不存在: {manual_path}") - return None - - print("\n选择模型:") - print("1. 使用最新模型") - print("2. 从列表中选择") - print("3. 手动输入模型路径") - - choice = input("请输入选择 (1-3): ").strip() + # 基础动作选择 + state_normalized = np.array(current_state).reshape(-1, *current_state.shape) / 255 + qs = model.predict(state_normalized, verbose=0)[0] + action = np.argmax(qs) + + previous_action = action + + # 执行动作 + new_state, reward, done, _ = env.step(action) + + # 更新状态 + current_state = new_state + total_reward += reward + step_count += 1 + + # 计算FPS + frame_time = time.time() - step_start + fps_counter.append(frame_time) + + # 每30步显示一次状态 + if step_count % 30 == 0: + fps = len(fps_counter)/sum(fps_counter) if fps_counter else 0 + velocity = env.vehicle.get_velocity() + speed_kmh = 3.6 * np.linalg.norm([velocity.x, velocity.y, velocity.z]) + + status = "✅" if reward > 0 else "⚠️" if reward < -1 else "➡️" + + print(f"{status} 步数: {step_count:4d} | FPS: {fps:4.1f} | " + f"速度: {speed_kmh:5.1f} km/h | 奖励: {reward:6.2f} | 累计: {total_reward:7.2f}") + + if done: + break - if choice == "1": - # 使用最新模型 - latest_model = find_latest_model(model_dir) - if latest_model: - print(f"选择最新模型: {os.path.basename(latest_model)}") - return latest_model - else: - print("无法找到最新模型") - return None + # 清理环境 + env.cleanup_actors() - elif choice == "2": - # 从列表中选择 - if not model_files: - print("没有可用的模型文件") - return None - - try: - index = int(input(f"请输入模型编号 (1-{len(model_files)}): ").strip()) - if 1 <= index <= len(model_files): - selected_model = model_files[index-1] - print(f"选择模型: {os.path.basename(selected_model)}") - return selected_model - else: - print("无效的编号") - return None - except ValueError: - print("无效的输入") - return None + # 判断结果 + success = total_reward > 5 + result = "成功" if success else "失败" - elif choice == "3": - # 手动输入路径 - manual_path = input("请输入模型完整路径: ").strip() - if os.path.exists(manual_path): - return manual_path - else: - print(f"错误: 文件不存在: {manual_path}") - return None + print(f"\nEpisode {episode_num} 结果: {result}") + print(f"总步数: {step_count}, 总奖励: {total_reward:.2f}") - else: - print("无效的选择，将使用最新模型") - latest_model = find_latest_model(model_dir) - if latest_model: - print(f"使用最新模型: {os.path.basename(latest_model)}") - return latest_model - else: - print("无法找到最新模型") - return None + return success, total_reward, step_count -if __name__ == '__main__': - # GPU内存配置 - gpu_options = tf.compat.v1.GPUOptions(per_process_gpu_memory_fraction=MEMORY_FRACTION) - tf.compat.v1.keras.backend.set_session(tf.compat.v1.Session(config=tf.compat.v1.ConfigProto(gpu_options=gpu_options))) - - # 选择模型 - print("="*60) - print("模型选择") - print("="*60) - - # 使用交互式选择模型 - MODEL_PATH = select_model_interactively() +def load_model_with_fallback(model_path): + """加载模型，支持多种格式和回退机制""" + print(f"尝试加载模型: {model_path}") + + # 定义自定义层 + custom_objects = { + 'Add': tf.keras.layers.Add, + 'Subtract': tf.keras.layers.Subtract, + 'Lambda': tf.keras.layers.Lambda, + 'Multiply': tf.keras.layers.Multiply + } - if MODEL_PATH is None: - print("无法加载模型，程序退出") - exit(1) - - # 加载训练好的模型 - print(f"\n加载模型: {MODEL_PATH}") try: - model = load_model(MODEL_PATH, custom_objects={'Add': tf.keras.layers.Add, - 'Subtract': tf.keras.layers.Subtract, - 'Lambda': tf.keras.layers.Lambda}) - except Exception as e: - print(f"使用自定义对象加载失败，尝试标准加载: {e}") + # 尝试加载完整模型 + model = load_model(model_path, custom_objects=custom_objects) + print(f"✅ 模型加载成功 (使用自定义层)") + return model + except Exception as e1: + print(f"使用自定义层加载失败: {e1}") try: - model = load_model(MODEL_PATH) + # 尝试不加载自定义层 + model = load_model(model_path) + print(f"✅ 模型加载成功 (基础加载)") + return model except Exception as e2: - print(f"加载模型失败: {e2}") - print("请检查模型文件是否完整，或尝试其他模型") - exit(1) + print(f"基础加载失败: {e2}") + + # 尝试使用 tf.keras.models.load_model 的不同参数 + try: + model = tf.keras.models.load_model( + model_path, + compile=False, + custom_objects=custom_objects + ) + print(f"✅ 模型加载成功 (不编译)") + return model + except Exception as e3: + print(f"所有加载尝试失败: {e3}") + raise ValueError(f"无法加载模型: {model_path}") + + +def comprehensive_model_evaluation(model_path, num_episodes=5): + """综合模型评估""" + print(f"\n{'='*60}") + print(f"开始综合模型评估") + print(f"模型路径: {model_path}") + print(f"测试轮次: {num_episodes}") + print(f"{'='*60}") + + # GPU配置 + gpu_options = tf.compat.v1.GPUOptions(per_process_gpu_memory_fraction=MEMORY_FRACTION) + tf.compat.v1.keras.backend.set_session( + tf.compat.v1.Session(config=tf.compat.v1.ConfigProto(gpu_options=gpu_options))) - print("模型加载成功!") - print(f"模型架构: {model.layers[-1].name}") + # 加载模型 + model = load_model_with_fallback(model_path) - # 检查是否是Dueling DQN - is_dueling = any('value' in layer.name or 'advantage' in layer.name for layer in model.layers) - print(f"模型类型: {'Dueling DQN' if is_dueling else 'Standard DQN'}") - - # 创建测试环境 + # 创建环境 env = CarEnv() - env.SHOW_CAM = False # 关闭小窗口预览 - - # 性能统计 - fps_counter = deque(maxlen=60) - episode_rewards = [] - episode_lengths = [] - success_count = 0 - total_episodes = 0 - - # 初始化预测 + env.SHOW_CAM = False + + # 预热模型 + print("预热模型...") model.predict(np.ones((1, env.im_height, env.im_width, 3)), verbose=0) - - print("\n" + "="*60) - print("开始测试改进的DQN模型!") - print("="*60) - print("请查看CARLA窗口观看智能体运行...") - print("按Ctrl+C停止测试") - print(f"模型类型: {'Dueling DQN with PER' if is_dueling else 'Standard DQN'}") - print(f"使用模型: {os.path.basename(MODEL_PATH)}") - - # 循环测试多个episode - episode_count = 0 - previous_action = 1 # 初始动作为保持 + + # 运行测试 + results = { + 'successes': 0, + 'total_rewards': [], + 'episode_lengths': [], + 'start_time': time.time() + } try: - while True: - episode_count += 1 - total_episodes += 1 - print(f'\n{"="*40}') - print(f'开始第 {episode_count} 个测试轮次') - print(f'{"="*40}') - - # 重置环境并获取初始状态 - # 测试时使用正常难度（相当于训练的第3阶段） - current_state = env.reset(401) # 401表示使用正常难度 - env.collision_hist = [] # 重置碰撞历史 - - done = False - total_reward = 0 - step_count = 0 - max_steps = SECONDS_PER_EPISODE * 60 # 最大步数限制 - - # 单次episode内的循环 - while not done and step_count < max_steps: - # FPS计数开始 - step_start = time.time() - - # 基于当前观察空间预测动作（使用改进的安全版本） - action, qs, uncertainty = get_safe_action_improved( - model, current_state, env, previous_action - ) - previous_action = action - - # 执行环境步进 - new_state, reward, done, _ = env.step(action) - - # 更新当前状态 - current_state = new_state - total_reward += reward - step_count += 1 - - # 计算帧时间，更新FPS计数器 - frame_time = time.time() - step_start - fps_counter.append(frame_time) - - # 每20步打印一次详细信息 - if step_count % 20 == 0: - fps = len(fps_counter)/sum(fps_counter) if fps_counter else 0 - speed_vector = env.vehicle.get_velocity() - speed_kmh = 3.6 * np.linalg.norm([speed_vector.x, speed_vector.y, speed_vector.z]) - - print(f'轮次 {episode_count} | 步数: {step_count:3d} | FPS: {fps:4.1f} | ' - f'速度: {speed_kmh:4.1f} km/h | ' - f'动作: [{qs[0]:>5.2f}, {qs[1]:>5.2f}, {qs[2]:>5.2f}, {qs[3]:>5.2f}, {qs[4]:>5.2f}] {action} | ' - f'奖励: {reward:5.2f} | 累计: {total_reward:6.2f}') - - # 如果完成（碰撞等），结束当前episode - if done: - break - - # episode结束时显示结果 - result = "成功到达终点!" if reward > 5 else "发生碰撞或失败" - success = reward > 5 + for episode in range(1, num_episodes + 1): + success, reward, length = run_test_episode(model, env, episode, use_advanced_safety=True) if success: - success_count += 1 - print(f"✓ 第 {episode_count} 轮: {result}") - else: - print(f"✗ 第 {episode_count} 轮: {result}") - - print(f'总步数: {step_count} | 总奖励: {total_reward:.2f}') + results['successes'] += 1 + results['total_rewards'].append(reward) + results['episode_lengths'].append(length) - # 记录统计 - episode_rewards.append(total_reward) - episode_lengths.append(step_count) + # 短暂暂停 + time.sleep(1) - # 显示统计摘要 - if len(episode_rewards) >= 5: - avg_reward = np.mean(episode_rewards[-5:]) - avg_steps = np.mean(episode_lengths[-5:]) - success_rate = (success_count / 5) * 100 if len(episode_rewards) >= 5 else 0 - - print(f"\n最近5轮统计:") - print(f" 平均奖励: {avg_reward:.2f}") - print(f" 平均步数: {avg_steps:.1f}") - print(f" 成功率: {success_rate:.1f}%") - - env.cleanup_actors() - - # 短暂暂停后开始下一轮 - time.sleep(2) - except KeyboardInterrupt: print("\n测试被用户中断") + except Exception as e: + print(f"测试过程中发生错误: {e}") finally: - # 显示最终统计 - print("\n" + "="*60) - print("测试完成!") - print("="*60) + # 清理环境 + env.cleanup_actors() + + # 计算统计 + results['end_time'] = time.time() + results['total_time'] = results['end_time'] - results['start_time'] - if total_episodes > 0: - success_rate = (success_count / total_episodes) * 100 - avg_reward = np.mean(episode_rewards) if episode_rewards else 0 - avg_steps = np.mean(episode_lengths) if episode_lengths else 0 + if results['total_rewards']: + results['success_rate'] = results['successes'] / len(results['total_rewards']) * 100 + results['avg_reward'] = np.mean(results['total_rewards']) + results['avg_length'] = np.mean(results['episode_lengths']) + results['max_reward'] = max(results['total_rewards']) + results['min_reward'] = min(results['total_rewards']) + + # 显示评估报告 + print(f"\n{'='*60}") + print("综合评估报告") + print(f"{'='*60}") + print(f"测试轮次: {num_episodes}") + print(f"成功次数: {results['successes']}") + print(f"成功率: {results.get('success_rate', 0):.1f}%") + print(f"平均奖励: {results.get('avg_reward', 0):.2f}") + print(f"平均步数: {results.get('avg_length', 0):.1f}") + print(f"最佳表现: {results.get('max_reward', 0):.2f}") + print(f"最差表现: {results.get('min_reward', 0):.2f}") + print(f"总测试时间: {results.get('total_time', 0):.1f}秒") + print(f"模型路径: {model_path}") + + # 保存评估结果 + timestamp = time.strftime("%Y%m%d_%H%M%S") + model_name = os.path.basename(model_path).replace('.model', '') + eval_file = f"model_evaluation_{model_name}_{timestamp}.json" + + # 转换numpy类型为Python原生类型 + serializable_results = {} + for key, value in results.items(): + if isinstance(value, np.ndarray): + serializable_results[key] = value.tolist() + elif isinstance(value, np.generic): + serializable_results[key] = value.item() + else: + serializable_results[key] = value + + serializable_results['model_path'] = model_path + serializable_results['model_name'] = model_name + serializable_results['evaluation_date'] = timestamp + serializable_results['num_episodes'] = num_episodes + + with open(eval_file, 'w') as f: + json.dump(serializable_results, f, indent=2) + + print(f"\n评估结果已保存到: {eval_file}") + + return results + + +def interactive_model_selection(model_files): + """交互式模型选择""" + if not model_files: + print("❌ 未找到任何模型文件") + return None + + print(f"\n找到 {len(model_files)} 个模型文件:") + for i, file_path in enumerate(model_files): + filename = os.path.basename(file_path) + file_size = os.path.getsize(file_path) / (1024 * 1024) # MB + mod_time = time.strftime("%Y-%m-%d %H:%M", time.localtime(os.path.getmtime(file_path))) + print(f" {i+1}. {filename} ({file_size:.1f} MB, 修改于: {mod_time})") + + while True: + try: + choice = input(f"\n请选择模型 (1-{len(model_files)}) 或按回车选择最新模型: ").strip() + + if choice == "": + # 选择最新的模型 + selected = model_files[0] + print(f"选择最新的模型: {os.path.basename(selected)}") + return selected - print(f"总测试轮次: {total_episodes}") - print(f"成功次数: {success_count}") - print(f"成功率: {success_rate:.1f}%") - print(f"平均奖励: {avg_reward:.2f}") - print(f"平均步数: {avg_steps:.1f}") - print(f"模型类型: {'Dueling DQN with PER' if is_dueling else 'Standard DQN'}") - print(f"使用模型: {os.path.basename(MODEL_PATH)}") + choice_idx = int(choice) - 1 + if 0 <= choice_idx < len(model_files): + selected = model_files[choice_idx] + print(f"选择模型: {os.path.basename(selected)}") + return selected + else: + print(f"请输入 1 到 {len(model_files)} 之间的数字") + except ValueError: + print("请输入有效的数字") + except KeyboardInterrupt: + print("\n选择被用户中断") + return None + + +def main(): + """主函数 - 自动查找和测试模型""" + print(f"\n{'='*60}") + print("自动驾驶模型测试系统") + print(f"{'='*60}") + + # 自动查找模型文件 + print("\n正在搜索模型文件...") + model_files = find_model_files() + + if not model_files: + print("❌ 未找到任何模型文件 (.model)") + print("请确保:") + print(" 1. 已经训练过模型") + print(" 2. 模型文件保存在 'models' 目录中") + print(" 3. 模型文件扩展名为 .model") - # 清理环境 - print("\n清理环境...") - env.cleanup_actors() \ No newline at end of file + # 尝试搜索其他可能的扩展名 + for ext in [".h5", ".keras", ".tf"]: + alt_files = find_model_files(pattern=f"*{ext}") + if alt_files: + print(f"\n找到 {len(alt_files)} 个 {ext} 格式的模型文件") + model_files = alt_files + break + + if not model_files: + return + + # 交互式选择模型 + selected_model = interactive_model_selection(model_files) + + if not selected_model: + print("未选择模型，退出测试") + return + + # 开始测试 + comprehensive_model_evaluation(selected_model, num_episodes=3) + + +def quick_test(): + """快速测试 - 自动选择最佳模型并运行少量测试""" + print("\n正在执行快速测试...") + + # 查找模型 + model_files = find_model_files() + + if not model_files: + print("❌ 未找到模型文件") + return + + # 自动选择最佳模型 + selected_model = select_best_model(model_files) + + if not selected_model: + print("❌ 无法选择模型") + return + + print(f"自动选择模型: {os.path.basename(selected_model)}") + + # 运行1个episode进行快速测试 + comprehensive_model_evaluation(selected_model, num_episodes=1) + + +if __name__ == '__main__': + import argparse + + parser = argparse.ArgumentParser(description='自动驾驶模型测试') + parser.add_argument('--quick', action='store_true', help='快速测试模式') + parser.add_argument('--model', type=str, help='指定模型文件路径') + parser.add_argument('--episodes', type=int, default=3, help='测试轮次数量') + + args = parser.parse_args() + + if args.model: + # 使用指定的模型文件 + if os.path.exists(args.model): + print(f"使用指定模型: {args.model}") + comprehensive_model_evaluation(args.model, num_episodes=args.episodes) + else: + print(f"❌ 指定的模型文件不存在: {args.model}") + elif args.quick: + # 快速测试模式 + quick_test() + else: + # 交互式测试模式 + main() \ No newline at end of file diff --git a/src/Unmanned_vehicle_AD_DQN/main.py b/src/Unmanned_vehicle_AD_DQN/main.py index 96bc64a77b..de06a3dddc 100644 --- a/src/Unmanned_vehicle_AD_DQN/main.py +++ b/src/Unmanned_vehicle_AD_DQN/main.py @@ -21,6 +21,71 @@ from Model import * from Hyperparameters import * +def extended_reward_calculation(env, action, reward, done, step_info): + """ + 扩展的奖励计算函数，用于多目标优化 + """ + # 获取车辆状态 + vehicle_location = env.vehicle.get_location() + velocity = env.vehicle.get_velocity() + speed_kmh = 3.6 * math.sqrt(velocity.x**2 + velocity.y**2) + + # 计算多目标指标 + metrics = {} + + # 1. 安全性指标 + # 最近行人距离 + min_ped_distance = env.last_ped_distance if hasattr(env, 'last_ped_distance') else float('inf') + safety_score = 0 + if min_ped_distance < 100: + if min_ped_distance > 12: + safety_score = 10 # 非常安全 + elif min_ped_distance > 8: + safety_score = 7 # 安全 + elif min_ped_distance > 5: + safety_score = 3 # 警告 + elif min_ped_distance > 3: + safety_score = 1 # 危险 + else: + safety_score = 0 # 极危险 + + metrics['safety'] = safety_score + + # 2. 效率指标 + progress = (vehicle_location.x + 81) / 236.0 # 从-81到155 + efficiency_score = progress * 100 # 进度百分比 + metrics['efficiency'] = efficiency_score + + # 3. 舒适度指标 + # 转向平滑性 + if hasattr(env, 'last_action') and env.last_action in [3, 4]: + if env.same_steer_counter > 2: # 连续同向转向 + comfort_score = -2 # 不舒适 + else: + comfort_score = 2 # 舒适 + else: + comfort_score = 5 # 直行，最舒适 + + metrics['comfort'] = comfort_score + + # 4. 规则遵循指标 + # 速度是否在合理范围内 + if 20 <= speed_kmh <= 40: + rule_score = 1.0 + elif 15 <= speed_kmh < 20 or 40 < speed_kmh <= 45: + rule_score = 0.7 + else: + rule_score = 0.3 + + metrics['rule_following'] = rule_score + + # 5. 特殊事件 + metrics['collision'] = len(env.collision_history) > 0 + metrics['off_road'] = vehicle_location.x < -90 or abs(vehicle_location.y + 195) > 30 + metrics['dangerous_action'] = speed_kmh > 45 and action in [3, 4] # 高速急转 + + return metrics + if __name__ == '__main__': FPS = 60 # 帧率 ep_rewards = [-200] # 存储每轮奖励 @@ -38,10 +103,47 @@ # 创建模型保存目录 if not os.path.isdir('models'): os.makedirs('models') + + # 创建专家数据目录 + if not os.path.isdir('expert_data'): + os.makedirs('expert_data') - # 创建智能体和环境 - 启用Dueling DQN和PER - agent = DQNAgent(use_dueling=True, use_per=True) # 开启新功能 + # 创建智能体和环境 - 启用所有高级功能 + agent = DQNAgent( + use_dueling=True, + use_per=True, + use_curriculum=True, + use_multi_objective=True + ) + env = CarEnv() + + # 设置训练策略 + agent.setup_training_strategies(env) + + # 可选：使用模仿学习进行预训练 + use_imitation_pretraining = False # 设置为True启用模仿学习预训练 + + if use_imitation_pretraining: + print("=" * 60) + print("开始模仿学习预训练阶段") + print("=" * 60) + + # 检查是否有现有的专家数据 + expert_files = glob.glob("expert_data/*.pkl") + if expert_files: + # 使用最新的专家数据 + latest_expert = max(expert_files, key=os.path.getctime) + agent.imitation_manager.load_expert_data(latest_expert) + else: + # 收集新的专家数据 + print("未找到专家数据，开始收集...") + agent.imitation_manager.collect_expert_demonstration(env, num_episodes=5) + + # 使用行为克隆进行预训练 + agent.model = agent.imitation_manager.pretrain_with_behavioral_cloning(agent.model, epochs=15) + agent.target_model.set_weights(agent.model.get_weights()) + print("模仿学习预训练完成!") # 启动训练线程并等待训练初始化完成 trainer_thread = Thread(target=agent.train_in_loop, daemon=True) @@ -64,25 +166,57 @@ 'buffer_size': [] } + # 多目标统计 + multi_obj_stats = { + 'safety': [], + 'efficiency': [], + 'comfort': [], + 'rule_following': [] + } + + # 课程学习阶段记录 + curriculum_stages = [] + # 迭代训练轮次 epds = [] for episode in tqdm(range(1, EPISODES + 1), ascii=True, unit='episodes'): env.collision_hist = [] # 重置碰撞历史 agent.tensorboard.step = episode # 设置TensorBoard步数 - # 重置每轮统计 - 重置得分和步数 + # 应用课程学习配置 + if agent.curriculum_manager: + config = agent.curriculum_manager.get_current_config() + # 这里可以根据配置调整环境难度 + # 例如：调整行人数量、速度等 + print(f"课程学习 - 阶段 {agent.curriculum_manager.current_stage}: " + f"行人(十字路口={config['pedestrian_cross']}, 普通={config['pedestrian_normal']})") + curriculum_stages.append(agent.curriculum_manager.current_stage) + + # 重置每轮统计 score = 0 step = 1 + + # 多目标指标记录 + episode_metrics = { + 'safety': [], + 'efficiency': [], + 'comfort': [], + 'rule_following': [] + } # 重置环境并获取初始状态 current_state = env.reset(episode) - # 重置完成标志并开始迭代直到本轮结束 + # 重置完成标志 done = False episode_start = time.time() - # 单次episode内的最大步数限制 - max_steps_per_episode = SECONDS_PER_EPISODE * FPS + # 应用课程学习的最大步数限制 + if agent.curriculum_manager: + config = agent.curriculum_manager.get_current_config() + max_steps_per_episode = config['max_episode_steps'] + else: + max_steps_per_episode = SECONDS_PER_EPISODE * FPS # 仅在给定秒数内运行 while not done and step < max_steps_per_episode: @@ -92,20 +226,36 @@ # 从Q网络获取动作（利用） qs = agent.get_qs(current_state) action = np.argmax(qs) - print(f'动作: [{qs[0]:>5.2f}, {qs[1]:>5.2f}, {qs[2]:>5.2f}, {qs[3]:>5.2f}, {qs[4]:>5.2f}] {action}') + if episode % 20 == 0: # 减少打印频率 + print(f'动作: [{qs[0]:>5.2f}, {qs[1]:>5.2f}, {qs[2]:>5.2f}, {qs[3]:>5.2f}, {qs[4]:>5.2f}] {action}') else: - # 随机选择动作（探索）- 扩展为5个动作 + # 随机选择动作（探索） action = np.random.randint(0, 5) # 添加延迟以匹配60FPS time.sleep(1 / FPS) - # 更频繁的状态更新 - if step % 5 == 0: - new_state, reward, done, _ = env.step(action) + # 执行动作并获取结果 + new_state, reward, done, _ = env.step(action) + + # 计算多目标指标 + if agent.multi_objective_optimizer: + step_info = {'step': step, 'action': action} + metrics = extended_reward_calculation(env, action, reward, done, step_info) - score += reward # 累加奖励 - agent.update_replay_memory((current_state, action, reward, new_state, done)) # 更新经验回放 - current_state = new_state # 更新当前状态 + # 记录指标 + for key in episode_metrics: + if key in metrics: + episode_metrics[key].append(metrics[key]) + + # 使用多目标优化器计算综合奖励 + composite_reward = agent.multi_objective_optimizer.compute_composite_reward(metrics) + reward = composite_reward # 使用综合奖励 + + score += reward # 累加奖励 + + # 更新经验回放 + agent.update_replay_memory((current_state, action, reward, new_state, done)) + current_state = new_state # 更新当前状态 step += 1 @@ -115,56 +265,86 @@ # 本轮结束 - 销毁所有actor env.cleanup_actors() + # 计算本轮平均指标 + avg_metrics = {} + for key, values in episode_metrics.items(): + if values: + avg_metrics[key] = np.mean(values) + # 记录到统计中 + if key in multi_obj_stats: + multi_obj_stats[key].append(avg_metrics[key]) + + # 更新课程学习 + success = score > 5 # 成功完成的阈值 + if agent.curriculum_manager: + stage_changed = agent.curriculum_manager.update_stage(success, score) + if stage_changed: + print(f"课程学习阶段已更新: {agent.curriculum_manager.current_stage}") + + # 更新多目标优化器权重 + if agent.multi_objective_optimizer and episode % 10 == 0: + agent.multi_objective_optimizer.adjust_weights(avg_metrics) + if episode % 50 == 0: + print(agent.multi_objective_optimizer.get_performance_report()) + # 更新成功计数 - if score > 5: # 成功完成的阈值 + if success: success_count += 1 # 动态保存最佳模型 if score > best_score: best_score = score - agent.model.save(f'models/{MODEL_NAME}_best_{score:.2f}.model') + model_suffix = f"advanced_best_{score:.2f}" + agent.model.save(f'models/{MODEL_NAME}_{model_suffix}.model') + print(f"新的最佳模型已保存: 得分={score:.2f}") # 记录得分统计 scores.append(score) - avg_scores.append(np.mean(scores[-10:])) # 计算最近10轮平均分 + avg_scores.append(np.mean(scores[-10:]) if len(scores) >= 10 else np.mean(scores)) - # 记录PER缓冲区信息（如果使用PER） + # 记录PER缓冲区信息 if hasattr(agent, 'replay_buffer'): per_stats['buffer_size'].append(len(agent.replay_buffer)) - print(f"PER缓冲区大小: {len(agent.replay_buffer)}/{REPLAY_MEMORY_SIZE}") # 定期聚合统计信息 if not episode % AGGREGATE_STATS_EVERY or episode == 1: - average_reward = np.mean(scores[-AGGREGATE_STATS_EVERY:]) # 平均奖励 - min_reward = min(scores[-AGGREGATE_STATS_EVERY:]) # 最小奖励 - max_reward = max(scores[-AGGREGATE_STATS_EVERY:]) # 最大奖励 + average_reward = np.mean(scores[-AGGREGATE_STATS_EVERY:]) if len(scores) >= AGGREGATE_STATS_EVERY else np.mean(scores) + min_reward = min(scores[-AGGREGATE_STATS_EVERY:]) if len(scores) >= AGGREGATE_STATS_EVERY else min(scores) + max_reward = max(scores[-AGGREGATE_STATS_EVERY:]) if len(scores) >= AGGREGATE_STATS_EVERY else max(scores) # 添加PER统计到TensorBoard + stats_dict = { + 'reward_avg': average_reward, + 'reward_min': min_reward, + 'reward_max': max_reward, + 'epsilon': Hyperparameters.EPSILON + } + if hasattr(agent, 'replay_buffer'): avg_buffer = np.mean(per_stats['buffer_size'][-AGGREGATE_STATS_EVERY:]) if per_stats['buffer_size'] else 0 - agent.tensorboard.update_stats( - reward_avg=average_reward, - reward_min=min_reward, - reward_max=max_reward, - epsilon=Hyperparameters.EPSILON, - buffer_size=avg_buffer - ) - else: - agent.tensorboard.update_stats( - reward_avg=average_reward, - reward_min=min_reward, - reward_max=max_reward, - epsilon=Hyperparameters.EPSILON - ) + stats_dict['buffer_size'] = avg_buffer + + # 添加多目标指标 + if agent.multi_objective_optimizer: + for obj in ['safety', 'efficiency', 'comfort', 'rule_following']: + if multi_obj_stats[obj]: + recent_avg = np.mean(multi_obj_stats[obj][-AGGREGATE_STATS_EVERY:]) if len(multi_obj_stats[obj]) >= AGGREGATE_STATS_EVERY else np.mean(multi_obj_stats[obj]) + stats_dict[f'{obj}_score'] = recent_avg + + agent.tensorboard.update_stats(**stats_dict) # 保存模型，仅当最小奖励达到设定值时 if min_reward >= MIN_REWARD and (episode not in epds): - model_suffix = "dueling_per" if agent.use_dueling and agent.use_per else "baseline" - agent.model.save( - f'models/{MODEL_NAME}_{model_suffix}__{max_reward:_>7.2f}max_{average_reward:_>7.2f}avg_{min_reward:_>7.2f}min__{int(time.time())}.model') + model_suffix = f"advanced_{max_reward:_>7.2f}max_{average_reward:_>7.2f}avg_{min_reward:_>7.2f}min" + agent.model.save(f'models/{MODEL_NAME}_{model_suffix}__{int(time.time())}.model') epds.append(episode) - print(f'轮次: {episode}, 得分: {score:.2f}, 成功次数: {success_count}') + + # 打印训练信息 + info_str = f'轮次: {episode:3d}, 得分: {score:6.2f}, 成功: {success_count:3d}' + if agent.curriculum_manager: + info_str += f', 阶段: {agent.curriculum_manager.current_stage}' + print(info_str) # 衰减探索率 if Hyperparameters.EPSILON > Hyperparameters.MIN_EPSILON: @@ -181,33 +361,51 @@ final_avg_reward = np.mean(scores[-AGGREGATE_STATS_EVERY:] if len(scores) >= AGGREGATE_STATS_EVERY else scores) final_min_reward = min(scores[-AGGREGATE_STATS_EVERY:] if len(scores) >= AGGREGATE_STATS_EVERY else scores) - model_suffix = "dueling_per_final" if agent.use_dueling and agent.use_per else "baseline_final" + model_suffix = "advanced_final" agent.model.save( f'models/{MODEL_NAME}_{model_suffix}__{final_max_reward:_>7.2f}max_{final_avg_reward:_>7.2f}avg_{final_min_reward:_>7.2f}min__{int(time.time())}.model') + + # 保存训练统计数据 + training_stats = { + 'scores': scores, + 'avg_scores': avg_scores, + 'multi_obj_stats': multi_obj_stats, + 'curriculum_stages': curriculum_stages, + 'final_scores': { + 'max': final_max_reward, + 'avg': final_avg_reward, + 'min': final_min_reward + } + } + + stats_file = f'training_stats_{int(time.time())}.pkl' + with open(stats_file, 'wb') as f: + pickle.dump(training_stats, f) + print(f"训练统计数据已保存到: {stats_file}") # 绘制训练曲线 - fig, axes = plt.subplots(2, 2, figsize=(15, 10)) + fig, axes = plt.subplots(3, 2, figsize=(15, 15)) # 1. 得分曲线 - axes[0, 0].plot(scores, label='每轮得分', alpha=0.6) - axes[0, 0].plot(avg_scores, label='平均得分(最近10轮)', linewidth=2) + axes[0, 0].plot(scores, label='每轮得分', alpha=0.6, linewidth=1) + axes[0, 0].plot(avg_scores, label='平均得分(最近10轮)', linewidth=2, color='red') axes[0, 0].set_ylabel('得分') axes[0, 0].set_xlabel('训练轮次') - axes[0, 0].set_title('训练进度') + axes[0, 0].set_title('训练进度 - 得分曲线') axes[0, 0].legend() axes[0, 0].grid(True, alpha=0.3) # 2. 探索率衰减曲线 eps_values = [max(MIN_EPSILON, 1.0 * (EPSILON_DECAY ** i)) for i in range(len(scores))] - axes[0, 1].plot(eps_values, color='red') + axes[0, 1].plot(eps_values, color='red', linewidth=2) axes[0, 1].set_ylabel('探索率 (ε)') axes[0, 1].set_xlabel('训练轮次') axes[0, 1].set_title('探索率衰减曲线') axes[0, 1].grid(True, alpha=0.3) - # 3. PER缓冲区大小（如果使用PER） + # 3. PER缓冲区大小 if per_stats['buffer_size']: - axes[1, 0].plot(per_stats['buffer_size'], color='green') + axes[1, 0].plot(per_stats['buffer_size'], color='green', linewidth=2) axes[1, 0].axhline(y=REPLAY_MEMORY_SIZE, color='r', linestyle='--', alpha=0.5, label='最大容量') axes[1, 0].set_ylabel('缓冲区大小') axes[1, 0].set_xlabel('训练轮次') @@ -215,18 +413,58 @@ axes[1, 0].legend() axes[1, 0].grid(True, alpha=0.3) - # 4. 成功次数统计 + # 4. 课程学习阶段变化 + if curriculum_stages: + axes[1, 1].plot(curriculum_stages, color='purple', linewidth=2, drawstyle='steps-post') + axes[1, 1].set_ylabel('课程学习阶段') + axes[1, 1].set_xlabel('训练轮次') + axes[1, 1].set_title('课程学习阶段变化') + axes[1, 1].grid(True, alpha=0.3) + + # 5. 多目标指标 + if multi_obj_stats['safety']: + colors = ['blue', 'green', 'orange', 'purple'] + for i, (key, values) in enumerate(multi_obj_stats.items()): + if values: + # 计算滑动平均 + window = 10 + if len(values) >= window: + smoothed = np.convolve(values, np.ones(window)/window, mode='valid') + axes[2, 0].plot(range(len(smoothed)), smoothed, label=key, color=colors[i], alpha=0.7) + else: + axes[2, 0].plot(values, label=key, color=colors[i], alpha=0.7) + + axes[2, 0].set_ylabel('分数') + axes[2, 0].set_xlabel('训练轮次') + axes[2, 0].set_title('多目标优化指标') + axes[2, 0].legend() + axes[2, 0].grid(True, alpha=0.3) + + # 6. 成功率统计 success_rates = [] for i in range(len(scores)): window = scores[max(0, i-9):i+1] success_rate = sum(1 for s in window if s > 5) / len(window) * 100 success_rates.append(success_rate) - axes[1, 1].plot(success_rates, color='purple') - axes[1, 1].set_ylabel('成功率 (%)') - axes[1, 1].set_xlabel('训练轮次') - axes[1, 1].set_title('最近10轮成功率') - axes[1, 1].grid(True, alpha=0.3) + axes[2, 1].plot(success_rates, color='darkred', linewidth=2) + axes[2, 1].axhline(y=80, color='g', linestyle='--', alpha=0.5, label='目标成功率80%') + axes[2, 1].set_ylabel('成功率 (%)') + axes[2, 1].set_xlabel('训练轮次') + axes[2, 1].set_title('最近10轮成功率') + axes[2, 1].legend() + axes[2, 1].grid(True, alpha=0.3) + plt.suptitle('高级训练策略 - 综合训练报告', fontsize=16, fontweight='bold') plt.tight_layout() - plt.show() \ No newline at end of file + plt.show() + + print("\n" + "="*60) + print("训练完成!") + print("="*60) + print(f"最终统计:") + print(f" 总轮次: {EPISODES}") + print(f" 最佳得分: {best_score:.2f}") + print(f" 平均得分: {np.mean(scores):.2f}") + print(f" 成功率: {(success_count/EPISODES)*100:.1f}%") + print(f" 最终探索率: {Hyperparameters.EPSILON:.4f}") \ No newline at end of file