#!/usr/bin/env python3 """ 详细训练教程 Detailed Training Tutorial for Emotion and Physiological State Prediction Model 这个脚本演示了如何训练情绪与生理状态变化预测模型的完整流程： 1. 数据准备和探索 2. 数据预处理 3. 模型配置和创建 4. 训练过程监控 5. 模型评估和验证 6. 超参数调优 运行方式： python training_tutorial.py """ import sys import os from pathlib import Path import numpy as np import pandas as pd import torch import torch.nn as nn import torch.optim as optim from torch.utils.data import DataLoader, TensorDataset import matplotlib.pyplot as plt import seaborn as sns from typing import Dict, Any, List, Tuple import yaml import json # 添加项目根目录到Python路径 project_root = Path(__file__).parent.parent sys.path.insert(0, str(project_root)) from src.data.synthetic_generator import SyntheticDataGenerator from src.models.pad_predictor import PADPredictor from src.data.preprocessor import DataPreprocessor from src.utils.trainer import ModelTrainer from src.utils.logger import setup_logger from src.models.loss_functions import WeightedMSELoss from src.models.metrics import RegressionMetrics def main(): """主函数""" print("=" * 80) print("情绪与生理状态变化预测模型 - 详细训练教程") print("Emotion and Physiological State Prediction Model - Detailed Training Tutorial") print("=" * 80) # 设置日志 setup_logger(level='INFO') # 创建输出目录 output_dir = Path(project_root) / "examples" / "training_outputs" output_dir.mkdir(exist_ok=True) # 1. 数据准备和探索 print("\n1. 数据准备和探索") print("-" * 50) train_data, val_data, test_data = prepare_and_explore_data(output_dir) # 2. 数据预处理 print("\n2. 数据预处理") print("-" * 50) preprocessor = preprocess_data(train_data, val_data, test_data, output_dir) # 3. 模型配置和创建 print("\n3. 模型配置和创建") print("-" * 50) model = create_and_configure_model() # 4. 训练配置 print("\n4. 训练配置") print("-" * 50) training_config = configure_training() # 5. 模型训练 print("\n5. 模型训练") print("-" * 50) history = train_model(model, preprocessor, train_data, val_data, training_config, output_dir) # 6. 模型评估 print("\n6. 模型评估") print("-" * 50) evaluate_model(model, preprocessor, test_data, output_dir) # 7. 超参数调优示例 print("\n7. 超参数调优示例") print("-" * 50) demonstrate_hyperparameter_tuning(output_dir) print("\n" + "=" * 80) print("详细训练教程完成！") print("Detailed Training Tutorial Completed!") print("=" * 80) def prepare_and_explore_data(output_dir: Path) -> Tuple[Tuple, Tuple, Tuple]: """数据准备和探索""" print(" - 生成不同模式的训练数据...") # 创建数据生成器 generator = SyntheticDataGenerator(seed=42) # 生成不同模式的数据 patterns = ['stress', 'relaxation', 'excitement', 'calm'] pattern_weights = [0.3, 0.3, 0.2, 0.2] # 生成训练数据 generator.num_samples = 2000 train_features, train_labels = generator.generate_dataset_with_patterns( patterns=patterns, pattern_weights=pattern_weights ) # 生成验证数据 generator.num_samples = 500 generator.seed = 123 val_features, val_labels = generator.generate_data(add_noise=True, add_correlations=True) # 生成测试数据 generator.num_samples = 300 generator.seed = 456 test_features, test_labels = generator.generate_data(add_noise=True, add_correlations=True) print(f" - 数据集大小:") print(f" 训练集: {train_features.shape}") print(f" 验证集: {val_features.shape}") print(f" 测试集: {test_features.shape}") # 数据探索和可视化 print(" - 生成数据探索图表...") visualize_data_exploration( train_features, train_labels, val_features, val_labels, test_features, test_labels, output_dir ) # 保存原始数据 save_data_splits( (train_features, train_labels), (val_features, val_labels), (test_features, test_labels), output_dir ) return (train_features, train_labels), (val_features, val_labels), (test_features, test_labels) def visualize_data_exploration(train_features, train_labels, val_features, val_labels, test_features, test_labels, output_dir: Path): """可视化数据探索""" # 特征列名 feature_columns = [ 'user_pleasure', 'user_arousal', 'user_dominance', 'vitality', 'current_pleasure', 'current_arousal', 'current_dominance' ] label_columns = [ 'delta_pleasure', 'delta_arousal', 'delta_dominance', 'delta_pressure', 'confidence' ] # 创建DataFrame train_df = pd.DataFrame(train_features, columns=feature_columns) train_labels_df = pd.DataFrame(train_labels, columns=label_columns) # 1. 特征分布图 fig, axes = plt.subplots(2, 4, figsize=(16, 8)) fig.suptitle('特征分布', fontsize=16) for i, col in enumerate(feature_columns): row, col_idx = i // 4, i % 4 axes[row, col_idx].hist(train_df[col], bins=30, alpha=0.7, color='skyblue') axes[row, col_idx].set_title(col) axes[row, col_idx].set_xlabel('值') axes[row, col_idx].set_ylabel('频率') # 隐藏最后一个子图 axes[1, 3].set_visible(False) plt.tight_layout() plt.savefig(output_dir / 'feature_distribution.png', dpi=300, bbox_inches='tight') plt.close() # 2. 标签分布图 fig, axes = plt.subplots(2, 3, figsize=(15, 10)) fig.suptitle('标签分布', fontsize=16) for i, col in enumerate(label_columns): row, col_idx = i // 3, i % 3 axes[row, col_idx].hist(train_labels_df[col], bins=30, alpha=0.7, color='lightcoral') axes[row, col_idx].set_title(col) axes[row, col_idx].set_xlabel('值') axes[row, col_idx].set_ylabel('频率') # 隐藏最后一个子图 axes[1, 2].set_visible(False) plt.tight_layout() plt.savefig(output_dir / 'label_distribution.png', dpi=300, bbox_inches='tight') plt.close() # 3. 相关性热力图 full_df = pd.concat([train_df, train_labels_df], axis=1) correlation_matrix = full_df.corr() plt.figure(figsize=(12, 10)) sns.heatmap(correlation_matrix, annot=True, cmap='coolwarm', center=0, square=True, fmt='.2f', cbar_kws={'label': '相关系数'}) plt.title('特征和标签相关性热力图') plt.tight_layout() plt.savefig(output_dir / 'correlation_heatmap.png', dpi=300, bbox_inches='tight') plt.close() print(f" - 数据探索图表已保存到: {output_dir}") def save_data_splits(train_data, val_data, test_data, output_dir: Path): """保存数据分割""" feature_columns = [ 'user_pleasure', 'user_arousal', 'user_dominance', 'vitality', 'current_pleasure', 'current_arousal', 'current_dominance' ] label_columns = [ 'delta_pleasure', 'delta_arousal', 'delta_dominance', 'delta_pressure', 'confidence' ] # 保存训练数据 train_df = pd.DataFrame(train_data[0], columns=feature_columns) train_labels_df = pd.DataFrame(train_data[1], columns=label_columns) train_full = pd.concat([train_df, train_labels_df], axis=1) train_full.to_csv(output_dir / 'train_data.csv', index=False) # 保存验证数据 val_df = pd.DataFrame(val_data[0], columns=feature_columns) val_labels_df = pd.DataFrame(val_data[1], columns=label_columns) val_full = pd.concat([val_df, val_labels_df], axis=1) val_full.to_csv(output_dir / 'val_data.csv', index=False) # 保存测试数据 test_df = pd.DataFrame(test_data[0], columns=feature_columns) test_labels_df = pd.DataFrame(test_data[1], columns=label_columns) test_full = pd.concat([test_df, test_labels_df], axis=1) test_full.to_csv(output_dir / 'test_data.csv', index=False) def preprocess_data(train_data, val_data, test_data, output_dir: Path) -> DataPreprocessor: """数据预处理""" print(" - 创建数据预处理器...") # 创建预处理器 preprocessor = DataPreprocessor() print(" - 拟合预处理器...") # 在训练数据上拟合预处理器 preprocessor.fit(train_data[0], train_data[1]) print(" - 转换数据...") # 转换所有数据集 train_processed = preprocessor.transform(train_data[0], train_data[1]) val_processed = preprocessor.transform(val_data[0], val_data[1]) test_processed = preprocessor.transform(test_data[0], test_data[1]) print(" - 创建数据加载器...") # 创建数据加载器 def create_dataloader(data, batch_size=32, shuffle=True): features, labels = data dataset = TensorDataset( torch.FloatTensor(features), torch.FloatTensor(labels) ) return DataLoader(dataset, batch_size=batch_size, shuffle=shuffle) train_loader = create_dataloader(train_processed, batch_size=32, shuffle=True) val_loader = create_dataloader(val_processed, batch_size=32, shuffle=False) test_loader = create_dataloader(test_processed, batch_size=32, shuffle=False) # 保存预处理器 preprocessor_path = output_dir / 'preprocessor.pkl' preprocessor.save(str(preprocessor_path)) print(f" - 预处理器已保存到: {preprocessor_path}") # 显示预处理信息 print(f" - 预处理统计:") print(f" 训练集样本数: {len(train_loader.dataset)}") print(f" 验证集样本数: {len(val_loader.dataset)}") print(f" 测试集样本数: {len(test_loader.dataset)}") return preprocessor def create_and_configure_model() -> PADPredictor: """创建和配置模型""" print(" - 加载模型配置...") # 加载模型配置 config_path = Path(project_root) / "configs" / "model_config.yaml" with open(config_path, 'r', encoding='utf-8') as f: model_config = yaml.safe_load(f) print(" - 创建模型...") # 创建模型 model = PADPredictor( input_dim=model_config['dimensions']['input_dim'], output_dim=model_config['dimensions']['output_dim'], hidden_dims=[layer['size'] for layer in model_config['architecture']['hidden_layers']], dropout_rate=model_config['architecture']['hidden_layers'][0]['dropout'], weight_init=model_config['initialization']['weight_init'], bias_init=model_config['initialization']['bias_init'] ) # 显示模型信息 model_info = model.get_model_info() print(f" - 模型信息:") print(f" 模型类型: {model_info['model_type']}") print(f" 输入维度: {model_info['input_dim']}") print(f" 输出维度: {model_info['output_dim']}") print(f" 隐藏层: {model_info['hidden_dims']}") print(f" 总参数数: {model_info['total_parameters']}") print(f" 可训练参数数: {model_info['trainable_parameters']}") return model def configure_training() -> Dict[str, Any]: """配置训练参数""" print(" - 配置训练参数...") # 加载训练配置 config_path = Path(project_root) / "configs" / "training_config.yaml" with open(config_path, 'r', encoding='utf-8') as f: training_config = yaml.safe_load(f) # 自定义一些训练参数 config = { 'epochs': 100, 'learning_rate': 0.001, 'weight_decay': 1e-4, 'batch_size': 32, 'patience': 15, 'min_delta': 1e-6, 'save_best_only': True, 'save_dir': Path(project_root) / "examples" / "training_outputs" / "models" } print(f" - 训练配置:") print(f" 训练轮数: {config['epochs']}") print(f" 学习率: {config['learning_rate']}") print(f" 权重衰减: {config['weight_decay']}") print(f" 批次大小: {config['batch_size']}") print(f" 早停耐心值: {config['patience']}") return config def train_model(model: PADPredictor, preprocessor: DataPreprocessor, train_data: Tuple, val_data: Tuple, training_config: Dict[str, Any], output_dir: Path) -> Dict[str, List]: """训练模型""" print(" - 创建训练器...") # 创建训练器 trainer = ModelTrainer(model, preprocessor) print(" - 创建数据加载器...") # 创建数据加载器 def create_dataloader(data, batch_size=32, shuffle=True): features, labels = data processed_features, processed_labels = preprocessor.transform(features, labels) dataset = TensorDataset( torch.FloatTensor(processed_features), torch.FloatTensor(processed_labels) ) return DataLoader(dataset, batch_size=batch_size, shuffle=shuffle) train_loader = create_dataloader(train_data, batch_size=training_config['batch_size'], shuffle=True) val_loader = create_dataloader(val_data, batch_size=training_config['batch_size'], shuffle=False) print(" - 开始训练...") # 开始训练 history = trainer.train( train_loader=train_loader, val_loader=val_loader, config=training_config ) print(" - 训练完成，保存结果...") # 保存训练历史 history_path = output_dir / 'training_history.json' with open(history_path, 'w', encoding='utf-8') as f: json.dump(history, f, indent=2, ensure_ascii=False) # 绘制训练曲线 plot_training_curves(history, output_dir) print(f" - 训练历史已保存到: {history_path}") return history def plot_training_curves(history: Dict[str, List], output_dir: Path): """绘制训练曲线""" fig, axes = plt.subplots(2, 2, figsize=(15, 10)) fig.suptitle('训练过程监控', fontsize=16) # 损失曲线 axes[0, 0].plot(history['train_loss'], label='训练损失', color='blue') axes[0, 0].plot(history['val_loss'], label='验证损失', color='red') axes[0, 0].set_title('损失曲线') axes[0, 0].set_xlabel('轮数') axes[0, 0].set_ylabel('损失') axes[0, 0].legend() axes[0, 0].grid(True) # 学习率曲线 if 'learning_rate' in history: axes[0, 1].plot(history['learning_rate'], color='green') axes[0, 1].set_title('学习率变化') axes[0, 1].set_xlabel('轮数') axes[0, 1].set_ylabel('学习率') axes[0, 1].grid(True) # 验证指标曲线 if 'val_metrics' in history: metrics = history['val_metrics'][0].keys() for i, metric in enumerate(metrics): if i < 2: # 只显示前两个指标 row, col = 1, i metric_values = [m[metric] for m in history['val_metrics']] axes[row, col].plot(metric_values, label=metric, color=f'C{i+2}') axes[row, col].set_title(f'验证指标: {metric}') axes[row, col].set_xlabel('轮数') axes[row, col].set_ylabel(metric) axes[row, col].legend() axes[row, col].grid(True) plt.tight_layout() plt.savefig(output_dir / 'training_curves.png', dpi=300, bbox_inches='tight') plt.close() print(f" - 训练曲线已保存到: {output_dir / 'training_curves.png'}") def evaluate_model(model: PADPredictor, preprocessor: DataPreprocessor, test_data: Tuple, output_dir: Path): """评估模型""" print(" - 加载最佳模型...") # 加载最佳模型 best_model_path = output_dir / 'models' / 'best_model.pth' if best_model_path.exists(): model = PADPredictor.load_model(str(best_model_path)) print(" - 在测试集上评估...") # 创建测试数据加载器 features, labels = test_data processed_features, processed_labels = preprocessor.transform(features, labels) model.eval() with torch.no_grad(): predictions = model(torch.FloatTensor(processed_features)) # 计算指标 metrics_calculator = RegressionMetrics() metrics = metrics_calculator.calculate_all_metrics( torch.FloatTensor(processed_labels), predictions ) print(" - 测试集评估结果:") for metric_name, value in metrics.items(): if isinstance(value, (int, float)): print(f" {metric_name}: {value:.4f}") # 保存评估结果 eval_results = { 'test_metrics': {k: float(v) if isinstance(v, (int, float)) else str(v) for k, v in metrics.items()}, 'model_info': model.get_model_info() } eval_path = output_dir / 'evaluation_results.json' with open(eval_path, 'w', encoding='utf-8') as f: json.dump(eval_results, f, indent=2, ensure_ascii=False) print(f" - 评估结果已保存到: {eval_path}") # 可视化预测结果 visualize_predictions(processed_labels, predictions.cpu().numpy(), output_dir) def visualize_predictions(true_labels: np.ndarray, predictions: np.ndarray, output_dir: Path): """可视化预测结果""" label_names = ['ΔPleasure', 'ΔArousal', 'ΔDominance', 'ΔPressure', 'Confidence'] fig, axes = plt.subplots(2, 3, figsize=(15, 10)) fig.suptitle('预测结果可视化', fontsize=16) for i in range(5): row, col = i // 3, i % 3 # 散点图 axes[row, col].scatter(true_labels[:, i], predictions[:, i], alpha=0.6, s=20) # 理想预测线 min_val = min(true_labels[:, i].min(), predictions[:, i].min()) max_val = max(true_labels[:, i].max(), predictions[:, i].max()) axes[row, col].plot([min_val, max_val], [min_val, max_val], 'r--', alpha=0.8) axes[row, col].set_xlabel('真实值') axes[row, col].set_ylabel('预测值') axes[row, col].set_title(label_names[i]) axes[row, col].grid(True, alpha=0.3) # 计算R² r2 = 1 - np.sum((true_labels[:, i] - predictions[:, i])**2) / np.sum((true_labels[:, i] - true_labels[:, i].mean())**2) axes[row, col].text(0.05, 0.95, f'R² = {r2:.3f}', transform=axes[row, col].transAxes, bbox=dict(boxstyle='round', facecolor='wheat', alpha=0.8)) # 隐藏最后一个子图 axes[1, 2].set_visible(False) plt.tight_layout() plt.savefig(output_dir / 'prediction_visualization.png', dpi=300, bbox_inches='tight') plt.close() print(f" - 预测结果可视化已保存到: {output_dir / 'prediction_visualization.png'}") def demonstrate_hyperparameter_tuning(output_dir: Path): """演示超参数调优""" print(" - 演示不同学习率的训练效果...") # 生成小批量数据用于快速演示 generator = SyntheticDataGenerator(num_samples=200, seed=789) features, labels = generator.generate_data() # 预处理数据 preprocessor = DataPreprocessor() processed_features, processed_labels = preprocessor.transform(features, labels) dataset = TensorDataset( torch.FloatTensor(processed_features), torch.FloatTensor(processed_labels) ) # 分割数据 train_size = int(0.8 * len(dataset)) val_size = len(dataset) - train_size train_dataset, val_dataset = torch.utils.data.random_split(dataset, [train_size, val_size]) train_loader = DataLoader(train_dataset, batch_size=16, shuffle=True) val_loader = DataLoader(val_dataset, batch_size=16, shuffle=False) # 测试不同学习率 learning_rates = [0.01, 0.001, 0.0001] results = {} for lr in learning_rates: print(f" 测试学习率: {lr}") # 创建模型 model = PADPredictor(input_dim=7, output_dim=5, hidden_dims=[64, 32]) # 创建优化器 optimizer = optim.Adam(model.parameters(), lr=lr, weight_decay=1e-4) # 训练少量轮数 trainer = ModelTrainer(model, preprocessor) config = { 'epochs': 20, 'learning_rate': lr, 'weight_decay': 1e-4, 'patience': 5, 'save_best_only': False } history = trainer.train(train_loader, val_loader, config) # 记录最终损失 final_val_loss = history['val_loss'][-1] results[str(lr)] = final_val_loss print(f" 最终验证损失: {final_val_loss:.4f}") # 保存调优结果 tuning_results = { 'learning_rates': learning_rates, 'final_val_losses': results, 'best_lr': min(results.keys(), key=lambda k: results[k]) } tuning_path = output_dir / 'hyperparameter_tuning.json' with open(tuning_path, 'w', encoding='utf-8') as f: json.dump(tuning_results, f, indent=2, ensure_ascii=False) print(f" - 超参数调优结果已保存到: {tuning_path}") print(f" - 最佳学习率: {tuning_results['best_lr']}") if __name__ == "__main__": main()