Buckets:
tahamajs/Sysmem2_in_AI / ComputerAssignments /CA17_reinforcement_learning_math /scripts /visualize.py
| #!/usr/bin/env python3 | |
| import argparse | |
| import os | |
| import sys | |
| import json | |
| from pathlib import Path | |
| sys.path.append(str(Path(__file__).parent.parent / "src")) | |
| from src.visualization.metrics_visualizer import MetricsVisualizer | |
| from src.visualization.plot_utils import ( | |
| plot_training_curves, | |
| plot_reward_distribution, | |
| plot_learning_progress, | |
| plot_comprehensive_analysis, | |
| plot_method_comparison, | |
| ) | |
| from src.utils.logging_utils import setup_logging | |
| from loguru import logger | |
| def parse_args(): | |
| parser = argparse.ArgumentParser(description="Visualize RL-LLM experiment results") | |
| parser.add_argument( | |
| "--results_dir", | |
| type=str, | |
| required=True, | |
| help="Directory containing experiment results", | |
| ) | |
| parser.add_argument( | |
| "--output_dir", | |
| type=str, | |
| default="./visualizations", | |
| help="Output directory for visualizations", | |
| ) | |
| parser.add_argument( | |
| "--log_level", | |
| type=str, | |
| default="INFO", | |
| choices=["DEBUG", "INFO", "WARNING", "ERROR"], | |
| help="Logging level", | |
| ) | |
| parser.add_argument( | |
| "--plot_types", | |
| nargs="+", | |
| default=["training", "performance", "comparison", "summary"], | |
| choices=["training", "performance", "comparison", "summary", "all"], | |
| help="Types of plots to generate", | |
| ) | |
| parser.add_argument( | |
| "--show_plots", action="store_true", help="Display plots interactively" | |
| ) | |
| parser.add_argument("--dpi", type=int, default=300, help="DPI for saved plots") | |
| parser.add_argument( | |
| "--format", | |
| type=str, | |
| default="png", | |
| choices=["png", "pdf", "svg", "jpg"], | |
| help="Output format for plots", | |
| ) | |
| return parser.parse_args() | |
| def load_experiment_data(results_dir: str): | |
| logger.info(f"Loading experiment data from {results_dir}") | |
| data = {} | |
| result_files = [ | |
| "experiment_results.json", | |
| "training_history.json", | |
| "final_metrics.json", | |
| "baseline_results.json", | |
| "comparison_results.json", | |
| ] | |
| for filename in result_files: | |
| filepath = os.path.join(results_dir, filename) | |
| if os.path.exists(filepath): | |
| try: | |
| with open(filepath, "r") as f: | |
| data[filename.replace(".json", "")] = json.load(f) | |
| logger.info(f"Loaded {filename}") | |
| except Exception as e: | |
| logger.warning(f"Failed to load {filename}: {e}") | |
| for filename in os.listdir(results_dir): | |
| if filename.endswith(".json") and filename not in result_files: | |
| filepath = os.path.join(results_dir, filename) | |
| try: | |
| with open(filepath, "r") as f: | |
| data[filename.replace(".json", "")] = json.load(f) | |
| logger.info(f"Loaded {filename}") | |
| except Exception as e: | |
| logger.warning(f"Failed to load {filename}: {e}") | |
| return data | |
| def create_training_plots(data: dict, output_dir: str, show_plots: bool = False): | |
| logger.info("Creating training plots") | |
| if "training_history" in data: | |
| training_history = data["training_history"] | |
| if isinstance(training_history, list) and len(training_history) > 0: | |
| plot_path = os.path.join(output_dir, "training_curves.png") | |
| plot_training_curves( | |
| training_history, save_path=plot_path, show_plot=show_plots | |
| ) | |
| logger.info(f"Training curves saved to {plot_path}") | |
| if "final_results" in data: | |
| final_results = data["final_results"] | |
| if isinstance(final_results, list) and len(final_results) > 0: | |
| rewards = [r.get("reward", 0) for r in final_results] | |
| plot_path = os.path.join(output_dir, "reward_distribution.png") | |
| plot_reward_distribution( | |
| rewards, | |
| title="Final Performance Reward Distribution", | |
| save_path=plot_path, | |
| show_plot=show_plots, | |
| ) | |
| logger.info(f"Reward distribution saved to {plot_path}") | |
| if "baseline_results" in data and "final_results" in data: | |
| baseline_rewards = [r.get("reward", 0) for r in data["baseline_results"]] | |
| final_rewards = [r.get("reward", 0) for r in data["final_results"]] | |
| plot_path = os.path.join(output_dir, "learning_progress.png") | |
| plot_learning_progress( | |
| baseline_rewards, final_rewards, save_path=plot_path, show_plot=show_plots | |
| ) | |
| logger.info(f"Learning progress saved to {plot_path}") | |
| def create_performance_plots(data: dict, output_dir: str, show_plots: bool = False): | |
| logger.info("Creating performance plots") | |
| visualizer = MetricsVisualizer(output_dir) | |
| if "training_history" in data: | |
| training_history = data["training_history"] | |
| if isinstance(training_history, list) and len(training_history) > 0: | |
| plot_path = visualizer.plot_training_metrics( | |
| training_history, "training_metrics" | |
| ) | |
| logger.info(f"Training metrics plot saved to {plot_path}") | |
| if "baseline_results" in data and "final_results" in data: | |
| baseline_results = {"results": data["baseline_results"]} | |
| final_results = {"results": data["final_results"]} | |
| plot_path = visualizer.plot_performance_comparison( | |
| baseline_results, final_results, "performance_comparison" | |
| ) | |
| logger.info(f"Performance comparison saved to {plot_path}") | |
| def create_comparison_plots(data: dict, output_dir: str, show_plots: bool = False): | |
| logger.info("Creating comparison plots") | |
| if "comparison_results" in data: | |
| comparison_data = data["comparison_results"] | |
| methods_data = {} | |
| for method_name, method_results in comparison_data.items(): | |
| if isinstance(method_results, dict) and "results" in method_results: | |
| results = method_results["results"] | |
| if isinstance(results, list) and len(results) > 0: | |
| rewards = [r.get("reward", 0) for r in results] | |
| success_rate = sum(1 for r in rewards if r > 0) / len(rewards) | |
| methods_data[method_name] = { | |
| "avg_reward": sum(rewards) / len(rewards), | |
| "success_rate": success_rate, | |
| "total_reward": sum(rewards), | |
| } | |
| if methods_data: | |
| plot_path = os.path.join(output_dir, "method_comparison.png") | |
| plot_method_comparison( | |
| methods_data, save_path=plot_path, show_plot=show_plots | |
| ) | |
| logger.info(f"Method comparison saved to {plot_path}") | |
| def create_summary_plots(data: dict, output_dir: str, show_plots: bool = False): | |
| logger.info("Creating summary plots") | |
| if any( | |
| key in data for key in ["training_history", "baseline_results", "final_results"] | |
| ): | |
| experiment_data = { | |
| "config": data.get("config", {}), | |
| "training_history": data.get("training_history", []), | |
| "baseline_results": data.get("baseline_results", []), | |
| "final_results": data.get("final_results", []), | |
| "baseline_metrics": data.get("baseline_metrics", {}), | |
| "final_metrics": data.get("final_metrics", {}), | |
| "timestamp": data.get("timestamp", "Unknown"), | |
| } | |
| plot_path = os.path.join(output_dir, "comprehensive_analysis.png") | |
| plot_comprehensive_analysis( | |
| experiment_data, save_path=plot_path, show_plot=show_plots | |
| ) | |
| logger.info(f"Comprehensive analysis saved to {plot_path}") | |
| visualizer = MetricsVisualizer(output_dir) | |
| if any( | |
| key in data for key in ["training_history", "baseline_results", "final_results"] | |
| ): | |
| experiment_data = { | |
| "config": data.get("config", {}), | |
| "training_history": data.get("training_history", []), | |
| "training_summary": data.get("training_summary", {}), | |
| "final_metrics": data.get("final_metrics", {}), | |
| "improvement": data.get("improvement", {}), | |
| "timestamp": data.get("timestamp", "Unknown"), | |
| } | |
| plot_path = visualizer.plot_experiment_summary( | |
| experiment_data, "experiment_summary" | |
| ) | |
| logger.info(f"Experiment summary saved to {plot_path}") | |
| def main(): | |
| args = parse_args() | |
| log_file = os.path.join(args.output_dir, "visualization.log") | |
| os.makedirs(args.output_dir, exist_ok=True) | |
| setup_logging(log_file, args.log_level) | |
| logger.info("Starting RL-LLM visualization") | |
| logger.info(f"Arguments: {args}") | |
| try: | |
| data = load_experiment_data(args.results_dir) | |
| if not data: | |
| logger.error("No experiment data found") | |
| return | |
| logger.info(f"Loaded {len(data)} data files") | |
| if "all" in args.plot_types or "training" in args.plot_types: | |
| create_training_plots(data, args.output_dir, args.show_plots) | |
| if "all" in args.plot_types or "performance" in args.plot_types: | |
| create_performance_plots(data, args.output_dir, args.show_plots) | |
| if "all" in args.plot_types or "comparison" in args.plot_types: | |
| create_comparison_plots(data, args.output_dir, args.show_plots) | |
| if "all" in args.plot_types or "summary" in args.plot_types: | |
| create_summary_plots(data, args.output_dir, args.show_plots) | |
| logger.info("=" * 50) | |
| logger.info("VISUALIZATION SUMMARY") | |
| logger.info("=" * 50) | |
| logger.info(f"Results directory: {args.results_dir}") | |
| logger.info(f"Output directory: {args.output_dir}") | |
| logger.info(f"Plots generated: {len(os.listdir(args.output_dir))}") | |
| logger.info(f"Data files processed: {len(data)}") | |
| plot_files = [ | |
| f | |
| for f in os.listdir(args.output_dir) | |
| if f.endswith((".png", ".pdf", ".svg", ".jpg")) | |
| ] | |
| if plot_files: | |
| logger.info("Generated plot files:") | |
| for plot_file in sorted(plot_files): | |
| logger.info(f" - {plot_file}") | |
| except Exception as e: | |
| logger.error(f"Visualization failed: {e}") | |
| raise | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 10.3 kB
- Xet hash:
- 2c311d7a96c9a76ebf20664c116f9996f35544e99d00fafeb5daae91becfac23
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.