rllib/agents/trainer.py
def step(self) -> ResultDict:
        """Implements the main `Trainer.train()` logic.

        Takes n attempts to perform a single training step. Thereby
        catches RayErrors resulting from worker failures. After n attempts,
        fails gracefully.

        Override this method in your Trainer sub-classes if you would like to
        handle worker failures yourself. Otherwise, override
        `self.step_attempt()` to keep the n attempts (catch worker failures).

        Returns:
            The results dict with stats/infos on sampling, training,
            and - if required - evaluation.
        """
        step_attempt_results = None

        with self._step_context() as step_ctx:
            while not step_ctx.should_stop(step_attempt_results):
                # Try to train one step.
                try:
                    step_attempt_results = self.step_attempt()
                # @ray.remote RolloutWorker failure.
                except RayError as e:
                    # Try to recover w/o the failed worker.
                    if (
                        self.config["ignore_worker_failures"]
                        or self.config["recreate_failed_workers"]
                    ):
                        logger.exception("Error in train call, attempting to recover")
                        self.try_recover_from_step_attempt()
                    # Error out.
                    else:
                        logger.warning(
                            "Worker crashed during call to `step_attempt()`. "
                            "To try to continue training without failed "
                            "worker(s), set `ignore_worker_failures=True`. "
                            "To try to recover the failed worker(s), set "
                            "`recreate_failed_workers=True`."
                        )
                        raise e
                # Any other exception.
                except Exception as e:
                    # Allow logs messages to propagate.
                    time.sleep(0.5)
                    raise e

        result = step_attempt_results

        if hasattr(self, "workers") and isinstance(self.workers, WorkerSet):
            # Sync filters on workers.
            self._sync_filters_if_needed(self.workers)

            # Collect worker metrics.
            if self.config["_disable_execution_plan_api"]:
                result = self._compile_step_results(
                    step_ctx=step_ctx,
                    step_attempt_results=step_attempt_results,
                )

        # Check `env_task_fn` for possible update of the env's task.
        if self.config["env_task_fn"] is not None:
            if not callable(self.config["env_task_fn"]):
                raise ValueError(
                    "`env_task_fn` must be None or a callable taking "
                    "[train_results, env, env_ctx] as args!"
                )

            def fn(env, env_context, task_fn):
                new_task = task_fn(result, env, env_context)
                cur_task = env.get_task()
                if cur_task != new_task:
                    env.set_task(new_task)

            fn = functools.partial(fn, task_fn=self.config["env_task_fn"])
            self.workers.foreach_env_with_context(fn)

        return result
Implements the main `Trainer.train()` logic.

        Takes n attempts to perform a single training step. Thereby
        catches RayErrors resulting from worker failures. After n attempts,
        fails gracefully.

        Override this method in your Trainer sub-classes if you would like to
        handle worker failures yourself. Otherwise, override
        `self.step_attempt()` to keep the n attempts (catch worker failures).

        Returns:
            The results dict with stats/infos on sampling, training,
            and - if required - evaluation.
        
Implements the main `Trainer.train()` logic.
Takes n attempts to perform a single training step. Thereby
catches RayErrors resulting from worker failures. After n attempts,
fails gracefully.

Override this method in your Trainer sub-classes if you would like to
handle worker failures yourself. Otherwise, override
`self.step_attempt()` to keep the n attempts (catch worker failures).
Implements the main ` Trainer . train () ` logic . Takes n attempts to perform a single training step . Thereby catches RayErrors resulting from worker failures . After n attempts fails gracefully . Override this method in your Trainer sub - classes if you would like to handle worker failures yourself . Otherwise override ` self . step_attempt () ` to keep the n attempts ( catch worker failures ) .
def step(self) -> ResultDict:
        step_attempt_results = None
        with self._step_context() as step_ctx:
            while not step_ctx.should_stop(step_attempt_results):
                try:
                    step_attempt_results = self.step_attempt()
                except RayError as e:
                    if (
                        self.config["ignore_worker_failures"]
                        or self.config["recreate_failed_workers"]
                    ):
                        logger.exception("Error in train call, attempting to recover")
                        self.try_recover_from_step_attempt()
                    else:
                        logger.warning(
                            "Worker crashed during call to `step_attempt()`. "
                            "To try to continue training without failed "
                            "worker(s), set `ignore_worker_failures=True`. "
                            "To try to recover the failed worker(s), set "
                            "`recreate_failed_workers=True`."
                        )
                        raise e
                except Exception as e:
                    time.sleep(0.5)
                    raise e
        result = step_attempt_results
        if hasattr(self, "workers") and isinstance(self.workers, WorkerSet):
            self._sync_filters_if_needed(self.workers)
            if self.config["_disable_execution_plan_api"]:
                result = self._compile_step_results(
                    step_ctx=step_ctx,
                    step_attempt_results=step_attempt_results,
                )
        if self.config["env_task_fn"] is not None:
            if not callable(self.config["env_task_fn"]):
                raise ValueError(
                    "`env_task_fn` must be None or a callable taking "
                    "[train_results, env, env_ctx] as args!"
                )
            def fn(env, env_context, task_fn):
                new_task = task_fn(result, env, env_context)
                cur_task = env.get_task()
                if cur_task != new_task:
                    env.set_task(new_task)
            fn = functools.partial(fn, task_fn=self.config["env_task_fn"])
            self.workers.foreach_env_with_context(fn)
        return result
def step ( self ) -> ResultDict : step_attempt_results = None with self . _step_context ( ) as step_ctx : while not step_ctx . should_stop ( step_attempt_results ) : try : step_attempt_results = self . step_attempt ( ) except RayError as e : if ( self . config [ "ignore_worker_failures" ] or self . config [ "recreate_failed_workers" ] ) : logger . exception ( "Error in train call, attempting to recover" ) self . try_recover_from_step_attempt ( ) else : logger . warning ( "Worker crashed during call to `step_attempt()`. " "To try to continue training without failed " "worker(s), set `ignore_worker_failures=True`. " "To try to recover the failed worker(s), set " "`recreate_failed_workers=True`." ) raise e except Exception as e : time . sleep ( 0.5 ) raise e result = step_attempt_results if hasattr ( self , "workers" ) and isinstance ( self . workers , WorkerSet ) : self . _sync_filters_if_needed ( self . workers ) if self . config [ "_disable_execution_plan_api" ] : result = self . _compile_step_results ( step_ctx = step_ctx , step_attempt_results = step_attempt_results , ) if self . config [ "env_task_fn" ] is not None : if not callable ( self . config [ "env_task_fn" ] ) : raise ValueError ( "`env_task_fn` must be None or a callable taking " "[train_results, env, env_ctx] as args!" ) def fn ( env , env_context , task_fn ) : new_task = task_fn ( result , env , env_context ) cur_task = env . get_task ( ) if cur_task != new_task : env . set_task ( new_task ) fn = functools . partial ( fn , task_fn = self . config [ "env_task_fn" ] ) self . workers . foreach_env_with_context ( fn ) return result
Implements the main `Trainer.train()` logic.
Implements the main ` Trainer . train () ` logic .
"""Implements the main `Trainer.train()` logic.

        Takes n attempts to perform a single training step. Thereby
        catches RayErrors resulting from worker failures. After n attempts,
        fails gracefully.

        Override this method in your Trainer sub-classes if you would like to
        handle worker failures yourself. Otherwise, override
        `self.step_attempt()` to keep the n attempts (catch worker failures).

        Returns:
            The results dict with stats/infos on sampling, training,
            and - if required - evaluation.
        """ # Try to train one step. # @ray.remote RolloutWorker failure. # Try to recover w/o the failed worker. # Error out. # Any other exception. # Allow logs messages to propagate. # Sync filters on workers. # Collect worker metrics. # Check `env_task_fn` for possible update of the env's task.
The results dict with stats/infos on sampling, training,
and - if required - evaluation.
The results dict with stats / infos on sampling training and - if required - evaluation .
self