code: def _build_multi_agent_batch(
      episode_id: int,
      batch_builder: _PolicyCollectorGroup,
      large_batch_threshold: int,
      multiple_episodes_in_batch: bool,
  ) -> MultiAgentBatch:
      """Build MultiAgentBatch from a dict of _PolicyCollectors.
  
      Args:
          env_steps: total env steps.
          policy_collectors: collected training SampleBatchs by policy.
  
      Returns:
          Always returns a sample batch in MultiAgentBatch format.
      """
      ma_batch = {}
      for pid, collector in batch_builder.policy_collectors.items():
          if collector.agent_steps <= 0:
              continue
  
          if (
              batch_builder.agent_steps > large_batch_threshold
              and log_once("large_batch_warning")
          ):
              logger.warning(
                  "More than {} observations in {} env steps for "
                  "episode {} ".format(
                      batch_builder.agent_steps,
                      batch_builder.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 multiple_episodes_in_batch
                      else ""
                  )
              )
  
          batch = collector.build()
  
          policy = collector.policy
  
          if policy.config.get("_enable_new_api_stack", False):
              # Before we send the collected batch back for training, we may need
              # to add a time dimension for the RLModule.
              seq_lens = batch.get(SampleBatch.SEQ_LENS)
              pad_batch_to_sequences_of_same_size(
                  batch=batch,
                  max_seq_len=policy.config["model"]["max_seq_len"],
                  shuffle=False,
                  batch_divisibility_req=getattr(
                      policy, "batch_divisibility_req", 1
                  ),
                  view_requirements=getattr(
                      policy, "view_requirements", None
                  ),
                  _enable_new_api_stack=True,
              )
              batch = policy.maybe_add_time_dimension(
                  batch, seq_lens=seq_lens, framework="np"
              )
  
          ma_batch[pid] = batch
  
      # Create the multi agent batch.
      return MultiAgentBatch(
          policy_batches=ma_batch, env_steps=batch_builder.env_steps
      )
  
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 1 for calling a global function with example usage: _build_multi_agent_batch(episode_id, batch_builder, large_batch_threshold, multiple_episodes_in_batch) and returns: MultiAgentBatch