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 episode_step(self, episode: Episode) -> None:
      episode_id = episode.episode_id
      # In the rase case that an "empty" step is taken at the beginning of
      # the episode (none of the agents has an observation in the obs-dict
      # and thus does not take an action), we have seen the episode before
      # and have to add it here to our registry.
      if episode_id not in self.episodes:
          self.episodes[episode_id] = episode
      else:
          assert episode is self.episodes[episode_id]
      self.episode_steps[episode_id] += 1
      episode.length += 1
  
      # In case of "empty" env steps (no agent is stepping), the builder
      # object may still be None.
      if episode.batch_builder:
          env_steps = episode.batch_builder.env_steps
          num_individual_observations = sum(
              c.agent_steps
              for c in episode.batch_builder.policy_collectors.values()
          )
  
          if (
              num_individual_observations > self.large_batch_threshold
              and log_once("large_batch_warning")
          ):
              logger.warning(
                  "More than {} observations in {} env steps for "
                  "episode {} ".format(
                      num_individual_observations, env_steps, episode_id
                  )
                  + "are buffered in the sampler. If this is more than you "
                  "expected, check that that you set a horizon on your "
                  "environment correctly and that it terminates at some "
                  "point. Note: In multi-agent environments, "
                  "`rollout_fragment_length` sets the batch size based on "
                  "(across-agents) environment steps, not the steps of "
                  "individual agents, which can result in unexpectedly "
                  "large batches."
                  + (
                      "Also, you may be waiting for your Env to "
                      "terminate (batch_mode=`complete_episodes`). Make sure "
                      "it does at some point."
                      if not self.multiple_episodes_in_batch
                      else ""
                  )
              )
  
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 2 for calling the SimpleListCollector obj's underlying member method code with example usage: obj.episode_step(episode) without return types