class_docstr: Tracks the current state of a (possibly multi-agent) episode.
class_docstr_tok: ["Tracks","the","current","state","of","a","(","possibly","multi-agent",")","episode","."]
docstr: Build and return currently collected training samples by policies.
  
  Clear agent collector states if this episode is done.
  
  Args:
      batch_builder: _PolicyCollectorGroup for saving the collected per-agent
          sample batches.
      is_done: If this episode is done (terminated or truncated).
      check_dones: Whether to make sure per-agent trajectories are actually done.
docstr_tok: ["Build","and","return","currently","collected","training","samples","by","policies",".","Clear","agent","collector","states","if","this","episode","is","done",".","Args",":","batch_builder",":","_PolicyCollectorGroup","for","saving","the","collected","per-agent","sample","batches",".","is_done",":","If","this","episode","is","done","(","terminated","or","truncated",")",".","check_dones",":","Whether","to","make","sure","per-agent","trajectories","are","actually","done","."]
code: def postprocess_episode(
      self,
      batch_builder: _PolicyCollectorGroup,
      is_done: bool = False,
      check_dones: bool = False,
  ) -> None:
      """Build and return currently collected training samples by policies.
  
      Clear agent collector states if this episode is done.
  
      Args:
          batch_builder: _PolicyCollectorGroup for saving the collected per-agent
              sample batches.
          is_done: If this episode is done (terminated or truncated).
          check_dones: Whether to make sure per-agent trajectories are actually done.
      """
      # TODO: (sven) Once we implement multi-agent communication channels,
      #  we have to resolve the restriction of only sending other agent
      #  batches from the same policy to the postprocess methods.
      # Build SampleBatches for the given episode.
      pre_batches = {}
      for agent_id, collector in self._agent_collectors.items():
          # Build only if there is data and agent is part of given episode.
          if collector.agent_steps == 0:
              continue
          pid = self.policy_for(agent_id)
          policy = self.policy_map[pid]
          pre_batch = collector.build_for_training(
              policy.view_requirements
          )
          pre_batches[agent_id] = (pid, policy, pre_batch)
  
      for agent_id, (pid, policy, pre_batch) in pre_batches.items():
          # Entire episode is said to be done.
          # Error if no DONE at end of this agent's trajectory.
          if (
              is_done
              and check_dones
              and not pre_batch.is_terminated_or_truncated()
          ):
              raise ValueError(
                  "Episode {} terminated for all agents, but we still "
                  "don't have a last observation for agent {} (policy "
                  "{}). ".format(
                      self.episode_id,
                      agent_id,
                      self.policy_for(agent_id),
                  )
                  + "Please ensure that you include the last observations "
                  "of all live agents when setting done[__all__] to "
                  "True."
              )
  
          # Skip a trajectory's postprocessing (and thus using it for training),
          # if its agent's info exists and contains the training_enabled=False
          # setting (used by our PolicyClients).
          if not self._last_infos.get(agent_id, {}).get(
              "training_enabled", True
          ):
              continue
  
          if (
              not pre_batch.is_single_trajectory()
              or len(np.unique(pre_batch[SampleBatch.EPS_ID])) > 1
          ):
              raise ValueError(
                  "Batches sent to postprocessing must only contain steps "
                  "from a single trajectory.",
                  pre_batch,
              )
  
          if len(pre_batches) > 1:
              other_batches = pre_batches.copy()
              del other_batches[agent_id]
          else:
              other_batches = {}
  
          # Call the Policy's Exploration's postprocess method.
          post_batch = pre_batch
          if getattr(policy, "exploration", None) is not None:
              policy.exploration.postprocess_trajectory(
                  policy, post_batch, policy.get_session()
              )
          post_batch.set_get_interceptor(None)
          post_batch = policy.postprocess_trajectory(
              post_batch, other_batches, self
          )
  
          from ray.rllib.evaluation.rollout_worker import (
              get_global_worker,
          )
  
          self.callbacks.on_postprocess_trajectory(
              worker=get_global_worker(),
              episode=self,
              agent_id=agent_id,
              policy_id=pid,
              policies=self.policy_map,
              postprocessed_batch=post_batch,
              original_batches=pre_batches,
          )
  
          # Append post_batch for return.
          if pid not in batch_builder.policy_collectors:
              batch_builder.policy_collectors[pid] = _PolicyCollector(
                  policy
              )
          batch_builder.policy_collectors[
              pid
          ].add_postprocessed_batch_for_training(
              post_batch, policy.view_requirements
          )
  
      batch_builder.agent_steps += self.active_agent_steps
      batch_builder.env_steps += self.active_env_steps
  
      # AgentCollector cleared.
      self.active_agent_steps = 0
      self.active_env_steps = 0
  
imports: import random
  from collections import defaultdict
  import numpy
  from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Tuple
  from ray.rllib.env.base_env import _DUMMY_AGENT_ID
  from ray.rllib.evaluation.collectors.simple_list_collector import _PolicyCollector, _PolicyCollectorGroup
  from ray.rllib.evaluation.collectors.agent_collector import AgentCollector
  from ray.rllib.policy.policy_map import PolicyMap
  from ray.rllib.policy.sample_batch import SampleBatch
  from ray.rllib.utils.annotations import DeveloperAPI
  from ray.rllib.utils.typing import AgentID, EnvID, EnvInfoDict, PolicyID, TensorType
label_desc: Use image node_id 8 for calling the EpisodeV2 obj's underlying member method code with example usage: obj.postprocess_episode(batch_builder, is_done, check_dones) without return types