(task: Task, step: Step)
| 56 | |
| 57 | |
| 58 | async def _generate_code(task: Task, step: Step) -> Step: |
| 59 | shared_deps = step.additional_properties["shared_deps"] |
| 60 | file_path = step.additional_properties["file_path"] |
| 61 | |
| 62 | code = await generate_code(task.input, shared_deps, file_path) |
| 63 | step.output = code |
| 64 | |
| 65 | write_file(os.path.join(Agent.get_workspace(task.task_id), file_path), code) |
| 66 | path = Path("./" + file_path) |
| 67 | await Agent.db.create_artifact( |
| 68 | task_id=task.task_id, |
| 69 | step_id=step.step_id, |
| 70 | agent_created=True, |
| 71 | relative_path=str(path.parent), |
| 72 | file_name=path.name, |
| 73 | ) |
| 74 | |
| 75 | return step |
| 76 | |
| 77 | |
| 78 | async def task_handler(task: Task) -> None: |
no test coverage detected