Run the task and return the states.
(
self,
task: Task,
batch_id: str,
repeat_times: int = 1,
run_id_base: int = 0,
collect_partial_runs: bool = True,
)
| 373 | return runner_state |
| 374 | |
| 375 | async def run_task( |
| 376 | self, |
| 377 | task: Task, |
| 378 | batch_id: str, |
| 379 | repeat_times: int = 1, |
| 380 | run_id_base: int = 0, |
| 381 | collect_partial_runs: bool = True, |
| 382 | ) -> Tuple[Status, bytes]: |
| 383 | """Run the task and return the states.""" |
| 384 | st = time.time() |
| 385 | try: |
| 386 | model_version = await self.model_wrapper.model_version_async |
| 387 | self.runner_state["model_version"] = model_version |
| 388 | self.logger.info( |
| 389 | f"Starting task: step={batch_id}, model_version={model_version}, repeat_times={repeat_times}, run_id_base={run_id_base}" |
| 390 | ) |
| 391 | execution_result = await self._run_task( |
| 392 | task, |
| 393 | repeat_times, |
| 394 | run_id_base, |
| 395 | collect_partial_runs=collect_partial_runs, |
| 396 | ) |
| 397 | model_version_after = await self.model_wrapper.model_version_async |
| 398 | exps = execution_result.experiences |
| 399 | if execution_result.status.completed_runs > 0: |
| 400 | assert exps is not None and len(exps) > 0, "An empty experience is generated" |
| 401 | # set eid for each experience |
| 402 | for exp in exps: |
| 403 | exp.eid.batch = task.batch_id |
| 404 | # keep exp.eid.task if it has been set before (e.g., in workflow) |
| 405 | if exp.eid.task == "": # "" is the default value |
| 406 | exp.eid.task = task.task_id |
| 407 | if not hasattr(exp, "info") or exp.info is None: |
| 408 | exp.info = {} |
| 409 | exp.info["model_version"] = model_version |
| 410 | exp.info["model_version_drift"] = model_version_after - model_version |
| 411 | exp.info["use_count"] = 0 |
| 412 | exp.info["task_index"] = task.index |
| 413 | |
| 414 | if not hasattr(exp, "metrics") or exp.metrics is None: |
| 415 | exp.metrics = {} |
| 416 | |
| 417 | status = execution_result.status |
| 418 | |
| 419 | if task.is_eval: |
| 420 | # If the task is an evaluation task, we do not record the experiences to the buffer |
| 421 | return status, b"" |
| 422 | else: |
| 423 | exp_payload = Experience.serialize_many(exps) |
| 424 | return status, exp_payload |
| 425 | |
| 426 | except Exception as e: |
| 427 | error_trace_back = traceback.format_exc() |
| 428 | self.logger.error(f"WorkflowRunner run task error: {e}\nTraceback:\n{error_trace_back}") |
| 429 | return ( |
| 430 | Status( |
| 431 | completed_runs=0, |
| 432 | total_runs=repeat_times, |