Init workflow from the task and run it.
(
self,
task: Task,
repeat_times: int,
run_id_base: int,
collect_partial_runs: bool = True,
)
| 281 | return False, [], None, error_trace_back.rstrip() |
| 282 | |
| 283 | async def _run_task( |
| 284 | self, |
| 285 | task: Task, |
| 286 | repeat_times: int, |
| 287 | run_id_base: int, |
| 288 | collect_partial_runs: bool = True, |
| 289 | ) -> RunnerExecutionResult: |
| 290 | """Init workflow from the task and run it.""" |
| 291 | if task.workflow.can_repeat: |
| 292 | workflow_instance = self._create_workflow_instance(task) |
| 293 | workflow_instance.set_repeat_times(repeat_times, run_id_base) |
| 294 | st = time.time() |
| 295 | await self.model_wrapper.clean_workflow_state() |
| 296 | self.runner_state["workflow_id"] = f"{task.batch_id}/{task.task_id}/{run_id_base}" |
| 297 | self.runner_state["terminate_time"] = None |
| 298 | self.runner_state["begin_time"] = st |
| 299 | exps = await self._run_workflow(workflow_instance) |
| 300 | et = time.time() |
| 301 | self.runner_state["terminate_time"] = et |
| 302 | # repeatable workflow cannot calculate run level metrics, we use experience level metrics directly |
| 303 | run_metrics = [exp.metrics for exp in exps if exp.metrics] |
| 304 | for metric in run_metrics: |
| 305 | metric["time/run_execution"] = et - st |
| 306 | return self._build_execution_result( |
| 307 | total_runs=repeat_times, |
| 308 | completed_runs=repeat_times, |
| 309 | metrics=run_metrics, |
| 310 | experiences=exps, |
| 311 | ) |
| 312 | else: |
| 313 | return await self.concurrent_run_fn( |
| 314 | task, |
| 315 | repeat_times, |
| 316 | run_id_base, |
| 317 | collect_partial_runs=collect_partial_runs, |
| 318 | ) |
| 319 | |
| 320 | async def _sequential_run( |
| 321 | self, |
no test coverage detected