class_docstr: Util to build SampleBatches for each policy in a multi-agent env.
  
  Input data is per-agent, while output data is per-policy. There is an M:N
  mapping between agents and policies. We retain one local batch builder
  per agent. When an agent is done, then its local batch is appended into the
  corresponding policy batch for the agent's policy.
code: def postprocess_episode(
      self,
      episode: Episode,
      is_done: bool = False,
      check_dones: bool = False,
      build: bool = False,
  ) -> Union[None, SampleBatch, MultiAgentBatch]:
      episode_id = episode.episode_id
      policy_collector_group = episode.batch_builder
  
      # Build SampleBatches for the given episode.
      pre_batches = {}
      for (
          eps_id,
          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 or eps_id != episode_id:
              continue
          pid = self.agent_key_to_policy_id[(eps_id, agent_id)]
          policy = self.policy_map[pid]
          pre_batch = collector.build_for_training(
              policy.view_requirements
          )
          pre_batches[agent_id] = (policy, pre_batch)
  
      # Apply reward clipping before calling postprocessing functions.
      if self.clip_rewards is True:
          for _, (_, pre_batch) in pre_batches.items():
              pre_batch["rewards"] = np.sign(pre_batch["rewards"])
      elif self.clip_rewards:
          for _, (_, pre_batch) in pre_batches.items():
              pre_batch["rewards"] = np.clip(
                  pre_batch["rewards"],
                  a_min=-self.clip_rewards,
                  a_max=self.clip_rewards,
              )
  
      post_batches = {}
      for agent_id, (_, 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(
                      episode_id,
                      agent_id,
                      self.agent_key_to_policy_id[
                          (episode_id, agent_id)
                      ],
                  )
                  + "Please ensure that you include the last observations "
                  "of all live agents when setting truncated[__all__] or "
                  "terminated[__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).
          last_info = episode.last_info_for(agent_id)
          if last_info and not last_info.get("training_enabled", True):
              if is_done:
                  agent_key = (episode_id, agent_id)
                  del self.agent_key_to_policy_id[agent_key]
                  del self.agent_collectors[agent_key]
              continue
  
          if len(pre_batches) > 1:
              other_batches = pre_batches.copy()
              del other_batches[agent_id]
          else:
              other_batches = {}
          pid = self.agent_key_to_policy_id[(episode_id, agent_id)]
          policy = self.policy_map[pid]
          if not pre_batch.is_single_trajectory():
              raise ValueError(
                  "Batches sent to postprocessing must be from a single trajectory! "
                  "TERMINATED & TRUNCATED need to be False everywhere, except the "
                  "last timestep, which can be either True or False for those keys)!",
                  pre_batch,
              )
          elif len(set(pre_batch[SampleBatch.EPS_ID])) > 1:
              episode_ids = set(pre_batch[SampleBatch.EPS_ID])
              raise ValueError(
                  "Batches sent to postprocessing must only contain steps "
                  "from a single episode! Your trajectory contains data from "
                  f"{len(episode_ids)} episodes ({list(episode_ids)}).",
                  pre_batch,
              )
          # Call the Policy's Exploration's postprocess method.
          post_batches[agent_id] = pre_batch
          if getattr(policy, "exploration", None) is not None:
              policy.exploration.postprocess_trajectory(
                  policy, post_batches[agent_id], policy.get_session()
              )
          post_batches[agent_id].set_get_interceptor(None)
          post_batches[agent_id] = policy.postprocess_trajectory(
              post_batches[agent_id], other_batches, episode
          )
  
      if log_once("after_post"):
          logger.info(
              "Trajectory fragment after postprocess_trajectory():\n\n{}\n".format(
                  summarize(post_batches)
              )
          )
  
      # Append into policy batches and reset.
      from ray.rllib.evaluation.rollout_worker import get_global_worker
  
      for agent_id, post_batch in sorted(post_batches.items()):
          agent_key = (episode_id, agent_id)
          pid = self.agent_key_to_policy_id[agent_key]
          policy = self.policy_map[pid]
          self.callbacks.on_postprocess_trajectory(
              worker=get_global_worker(),
              episode=episode,
              agent_id=agent_id,
              policy_id=pid,
              policies=self.policy_map,
              postprocessed_batch=post_batch,
              original_batches=pre_batches,
          )
  
          # Add the postprocessed SampleBatch to the policy collectors for
          # training.
          # PID may be a newly added policy. Just confirm we have it in our
          # policy map before proceeding with adding a new _PolicyCollector()
          # to the group.
          if pid not in policy_collector_group.policy_collectors:
              assert pid in self.policy_map
              policy_collector_group.policy_collectors[
                  pid
              ] = _PolicyCollector(policy)
          policy_collector_group.policy_collectors[
              pid
          ].add_postprocessed_batch_for_training(
              post_batch, policy.view_requirements
          )
  
          if is_done:
              del self.agent_key_to_policy_id[agent_key]
              del self.agent_collectors[agent_key]
  
      if policy_collector_group:
          env_steps = self.episode_steps[episode_id]
          policy_collector_group.env_steps += env_steps
          agent_steps = self.agent_steps[episode_id]
          policy_collector_group.agent_steps += agent_steps
  
      if is_done:
          del self.episode_steps[episode_id]
          del self.episodes[episode_id]
  
          if episode_id in self.agent_steps:
              del self.agent_steps[episode_id]
          else:
              assert (
                  len(pre_batches) == 0
              ), "Expected the batch to be empty since the episode_id is missing."
              # 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_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)
  
          # Make PolicyCollectorGroup available for more agent batches in
          # other episodes. Do not reset count to 0.
          if policy_collector_group:
              self.policy_collector_groups.append(
                  policy_collector_group
              )
      else:
          self.episode_steps[episode_id] = self.agent_steps[
              episode_id
          ] = 0
  
      # Build a MultiAgentBatch from the episode and return.
      if build:
          return self._build_multi_agent_batch(episode)
  
imports: import collections
  from gymnasium.spaces import Space
  import logging
  import numpy
  import tree
  from typing import Dict, List, Optional, Tuple, TYPE_CHECKING, Union
  from ray.rllib.env.base_env import _DUMMY_AGENT_ID
  from ray.rllib.evaluation.collectors.sample_collector import SampleCollector
  from ray.rllib.evaluation.collectors.agent_collector import AgentCollector
  from ray.rllib.evaluation.episode import Episode
  from ray.rllib.policy.policy import Policy
  from ray.rllib.policy.policy_map import PolicyMap
  from ray.rllib.policy.sample_batch import SampleBatch, MultiAgentBatch, concat_samples
  from ray.rllib.utils.annotations import override, PublicAPI
  from ray.rllib.utils.debug import summarize
  from ray.rllib.utils.framework import try_import_tf, try_import_torch
  from ray.rllib.utils.spaces.space_utils import get_dummy_batch_for_space
  from ray.rllib.utils.typing import AgentID, EpisodeID, EnvID, PolicyID, TensorType, ViewRequirementsDict
  from ray.util.debug import log_once
label_desc: Use image node_id 8 for calling the SimpleListCollector obj's underlying member method code with example usage: obj.postprocess_episode(episode, is_done, check_dones, build) and returns: self