Save experiment results to JSON file with real-time updates.
| 48 | |
| 49 | |
| 50 | class ExperimentResultSaver: |
| 51 | """Save experiment results to JSON file with real-time updates.""" |
| 52 | |
| 53 | def __init__(self, optimizer_type: str, benchmark_name: str, concurrency: int, total_tasks: int, model_name: str): |
| 54 | self.optimizer_type = optimizer_type |
| 55 | self.benchmark_name = benchmark_name |
| 56 | self.concurrency = concurrency |
| 57 | self.total_tasks = total_tasks |
| 58 | self.model_name = model_name |
| 59 | self.start_time = datetime.now() |
| 60 | |
| 61 | # Create results directory if it doesn't exist |
| 62 | self.results_dir = Path(__file__).parent / "workdir/results" |
| 63 | self.results_dir.mkdir(parents=True, exist_ok=True) |
| 64 | |
| 65 | # Generate filename with timestamp |
| 66 | timestamp = self.start_time.strftime("%Y-%m-%d_%H-%M-%S") |
| 67 | self.filename = f"{optimizer_type}_{benchmark_name}_{timestamp}.json" |
| 68 | self.filepath = self.results_dir / self.filename |
| 69 | |
| 70 | # Initialize results structure |
| 71 | self.results_data = { |
| 72 | "experiment_meta": { |
| 73 | "timestamp": self.start_time.isoformat() + "Z", |
| 74 | "optimizer": optimizer_type, |
| 75 | "benchmark": benchmark_name, |
| 76 | "concurrency": concurrency, |
| 77 | "total_tasks": total_tasks, |
| 78 | "model": model_name |
| 79 | }, |
| 80 | "results": [], |
| 81 | "summary": { |
| 82 | "completed_tasks": 0, |
| 83 | "correct_answers": 0, |
| 84 | "accuracy": 0.0, |
| 85 | "last_updated": self.start_time.isoformat() + "Z" |
| 86 | } |
| 87 | } |
| 88 | |
| 89 | # Save initial empty results |
| 90 | self._save_to_file() |
| 91 | |
| 92 | def add_task_result(self, task_data: Any, processing_time: float = None, |
| 93 | optimizer_data: Dict[str, Any] = None): |
| 94 | """Add a single task result and update the file.""" |
| 95 | _, answer = parse_agent_result(task_data.result) |
| 96 | |
| 97 | task_result = {"task_id": task_data.task_id, |
| 98 | "task_input": task_data.input, |
| 99 | "ground_truth": str(task_data.ground_truth), |
| 100 | "result": answer, |
| 101 | "reasoning": getattr(task_data, 'reasoning', ""), |
| 102 | "correct": getattr(task_data, 'result', "") == str(task_data.ground_truth), |
| 103 | "processing_time": processing_time, "reflection_process": { |
| 104 | "initial_reasoning": optimizer_data.get("initial_agent_reasoning", ""), |
| 105 | "initial_result": optimizer_data.get("initial_agent_result", ""), |
| 106 | "reflection_rounds": [] |
| 107 | }} |
no outgoing calls
no test coverage detected