Generate a single source file.
(
self,
file_spec: dict,
code_plan: dict,
blueprint: dict,
setup: dict,
)
| 425 | return self._normalize_code_plan(result if isinstance(result, dict) else {}) |
| 426 | |
| 427 | async def _generate_file( |
| 428 | self, |
| 429 | file_spec: dict, |
| 430 | code_plan: dict, |
| 431 | blueprint: dict, |
| 432 | setup: dict, |
| 433 | ) -> str: |
| 434 | """Generate a single source file.""" |
| 435 | code_analysis = setup.get("code_analysis", {}) |
| 436 | cloned_repos = setup.get("cloned_repos", []) |
| 437 | downloaded_resources = setup.get("downloaded_resources", []) |
| 438 | data_dir = setup.get("data_dir", "") |
| 439 | models_dir = setup.get("models_dir", "") |
| 440 | |
| 441 | resource_paths = self._format_resource_paths(downloaded_resources, data_dir, models_dir) |
| 442 | |
| 443 | # Find relevant reference code from cloned repos |
| 444 | reference_code = "" |
| 445 | for component in code_analysis.get("reusable_components", [])[:3]: |
| 446 | source_file = component.get("source_file", "") |
| 447 | if source_file and Path(source_file).exists(): |
| 448 | try: |
| 449 | content = Path(source_file).read_text(errors="replace")[:3000] |
| 450 | reference_code += f"\n# Reference from {source_file}:\n{content}\n" |
| 451 | except Exception as exc: |
| 452 | logger.debug("Failed to read reference code from %s: %s", source_file, exc) |
| 453 | if len(reference_code) > 8000: |
| 454 | reference_code = reference_code[:8000] |
| 455 | |
| 456 | system_prompt = ( |
| 457 | "You are a senior ML engineer writing production-quality research code. " |
| 458 | "Write COMPLETE, RUNNABLE Python code. No stubs, no TODOs, no placeholders. " |
| 459 | "The code must actually work when executed. " |
| 460 | "Use standard PyTorch patterns. Include proper error handling, " |
| 461 | "logging, and metric tracking. Save results to JSON/CSV files. " |
| 462 | "On ANY unhandled error in the main training script, call sys.exit(1) — " |
| 463 | "never let exceptions be silently caught with exit code 0." |
| 464 | ) |
| 465 | |
| 466 | all_files = [f.get("path", "") for f in code_plan.get("files", [])] |
| 467 | |
| 468 | user_prompt = f"""Write the complete code for: {file_spec.get('path', '')} |
| 469 | Description: {file_spec.get('description', '')} |
| 470 | |
| 471 | Project structure: {json.dumps(all_files)} |
| 472 | Method: {json.dumps(blueprint.get('proposed_method', {}), indent=2)[:1000]} |
| 473 | Datasets: {json.dumps(blueprint.get('datasets', []), indent=2)[:500]} |
| 474 | Metrics: {json.dumps(blueprint.get('metrics', []), indent=2)[:300]} |
| 475 | Experiment matrix: {json.dumps(blueprint.get('experiment_matrix', []), indent=2)[:2500]} |
| 476 | Required artifacts: {json.dumps(blueprint.get('required_artifacts', []), indent=2)} |
| 477 | Minimum success criteria: {json.dumps(blueprint.get('minimum_success_criteria', {}), indent=2)} |
| 478 | Dependencies: {json.dumps(code_plan.get('dependencies', []))} |
| 479 | Train command: {code_plan.get('train_command', 'python run_experiments.py --matrix configs/experiment_matrix.json --output results')} |
| 480 | |
| 481 | === ALREADY DOWNLOADED DATA & MODELS (use these exact paths) === |
| 482 | {resource_paths} |
| 483 | |
| 484 | {f'Reference code from existing repos:{reference_code}' if reference_code else ''} |
no test coverage detected