class_docstr: Collect experiences from user environment using Connectors.
docstr: Cleans up an episode that has finished.
  
  Args:
      env_id: Env ID.
      episode_or_exception: Instance of an episode if it finished successfully.
          Otherwise, the exception that was thrown,
docstr_tok: ["Cleans","up","an","episode","that","has","finished",".","Args",":","env_id",":","Env","ID",".","episode_or_exception",":","Instance","of","an","episode","if","it","finished","successfully",".","Otherwise",",","the","exception","that","was","thrown",","]
code: def end_episode(
      self,
      env_id: EnvID,
      episode_or_exception: Union[EpisodeV2, Exception],
  ):
      """Cleans up an episode that has finished.
  
      Args:
          env_id: Env ID.
          episode_or_exception: Instance of an episode if it finished successfully.
              Otherwise, the exception that was thrown,
      """
      # Signal the end of an episode, either successfully with an Episode or
      # unsuccessfully with an Exception.
      self._callbacks.on_episode_end(
          worker=self._worker,
          base_env=self._base_env,
          policies=self._worker.policy_map,
          episode=episode_or_exception,
          env_index=env_id,
      )
  
      # Call each (in-memory) policy's Exploration.on_episode_end
      # method.
      # Note: This may break the exploration (e.g. ParameterNoise) of
      # policies in the `policy_map` that have not been recently used
      # (and are therefore stashed to disk). However, we certainly do not
      # want to loop through all (even stashed) policies here as that
      # would counter the purpose of the LRU policy caching.
      for p in self._worker.policy_map.cache.values():
          if getattr(p, "exploration", None) is not None:
              p.exploration.on_episode_end(
                  policy=p,
                  environment=self._base_env,
                  episode=episode_or_exception,
                  tf_sess=p.get_session(),
              )
  
      if isinstance(episode_or_exception, EpisodeV2):
          episode = episode_or_exception
          if episode.total_agent_steps == 0:
              # if the key does not exist it means that throughout the episode all
              # observations were empty (i.e. there was no agent in the env)
              msg = (
                  f"Data from episode {episode.episode_id} does not show any agent "
                  f"interactions. Hint: Make sure for at least one timestep in the "
                  f"episode, env.step() returns non-empty values."
              )
              raise ValueError(msg)
  
      # Clean up the episode and batch_builder for this env id.
      if env_id in self._active_episodes:
          del self._active_episodes[env_id]
  
imports: from collections import defaultdict
  import logging
  import time
  import tree
  from typing import TYPE_CHECKING, Dict, Iterator, List, Optional, Set, Tuple, Union
  import numpy
  from ray.rllib.env.base_env import ASYNC_RESET_RETURN, BaseEnv
  from ray.rllib.env.external_env import ExternalEnvWrapper
  from ray.rllib.env.wrappers.atari_wrappers import MonitorEnv, get_wrapper_by_cls
  from ray.rllib.evaluation.collectors.simple_list_collector import _PolicyCollectorGroup
  from ray.rllib.policy.rnn_sequencing import pad_batch_to_sequences_of_same_size
  from ray.rllib.evaluation.episode_v2 import EpisodeV2
  from ray.rllib.evaluation.metrics import RolloutMetrics
  from ray.rllib.models.preprocessors import Preprocessor
  from ray.rllib.policy.policy import Policy
  from ray.rllib.policy.sample_batch import MultiAgentBatch, SampleBatch, concat_samples
  from ray.rllib.utils.annotations import DeveloperAPI
  from ray.rllib.utils.filter import Filter
  from ray.rllib.utils.numpy import convert_to_numpy
  from ray.rllib.utils.spaces.space_utils import unbatch, get_original_space
  from ray.rllib.utils.typing import ActionConnectorDataType, AgentConnectorDataType, AgentID, EnvActionType, EnvID, EnvInfoDict, EnvObsType, MultiAgentDict, MultiEnvDict, PolicyID, PolicyOutputType, SampleBatchType, StateBatches, TensorStructType
  from ray.util.debug import log_once
label_desc: Use image node_id 13 for calling the EnvRunnerV2 obj's underlying member method code with example usage: obj.end_episode(env_id, episode_or_exception) without return types