class_docstr: A replay buffer shard for multiagent setups.
  
  This buffer is meant to be run in parallel to distribute experiences
  across `num_shards` shards. Unlike simpler buffers, it holds a set of
  buffers - one for each policy ID.
docstr: Add a batch of experiences to the underlying buffer of a policy.
  
  If the storage unit is `timesteps`, cut the batch into timeslices
  before adding them to the appropriate buffer. Otherwise, let the
  underlying buffer decide how slice batches.
  
  Args:
      policy_id: ID of the policy that corresponds to the underlying
      buffer
      batch: SampleBatch to add to the underlying buffer
      ``**kwargs``: Forward compatibility kwargs.
docstr_tok: ["Add","a","batch","of","experiences","to","the","underlying","buffer","of","a","policy",".","If","the","storage","unit","is","`","timesteps","`",",","cut","the","batch","into","timeslices","before","adding","them","to","the","appropriate","buffer",".","Otherwise",",","let","the","underlying","buffer","decide","how","slice","batches",".","Args",":","policy_id",":","ID","of","the","policy","that","corresponds","to","the","underlying","buffer","batch",":","SampleBatch","to","add","to","the","underlying","buffer","``","*","*","kwargs","``",":","Forward","compatibility","kwargs","."]
code: def _add_to_underlying_buffer(
      self, policy_id: PolicyID, batch: SampleBatchType, **kwargs
  ) -> None:
      """Add a batch of experiences to the underlying buffer of a policy.
  
      If the storage unit is `timesteps`, cut the batch into timeslices
      before adding them to the appropriate buffer. Otherwise, let the
      underlying buffer decide how slice batches.
  
      Args:
          policy_id: ID of the policy that corresponds to the underlying
          buffer
          batch: SampleBatch to add to the underlying buffer
          ``**kwargs``: Forward compatibility kwargs.
      """
      # Merge kwargs, overwriting standard call arguments
      kwargs = merge_dicts_with_warning(
          self.underlying_buffer_call_args, kwargs
      )
  
      # For the storage unit `timesteps`, the underlying buffer will
      # simply store the samples how they arrive. For sequences and
      # episodes, the underlying buffer may split them itself.
      if self.storage_unit is StorageUnit.TIMESTEPS:
          timeslices = batch.timeslices(1)
      elif self.storage_unit is StorageUnit.SEQUENCES:
          timeslices = timeslice_along_seq_lens_with_overlap(
              sample_batch=batch,
              seq_lens=batch.get(SampleBatch.SEQ_LENS)
              if self.replay_sequence_override
              else None,
              zero_pad_max_seq_len=self.replay_sequence_length,
              pre_overlap=self.replay_burn_in,
              zero_init_states=self.replay_zero_init_states,
          )
      elif self.storage_unit == StorageUnit.EPISODES:
          timeslices = []
          for eps in batch.split_by_episode():
              if eps.get(SampleBatch.T)[0] == 0 and (
                  eps.get(SampleBatch.TERMINATEDS, [True])[-1]
                  or eps.get(SampleBatch.TRUNCATEDS, [False])[-1]
              ):
                  # Only add full episodes to the buffer
                  timeslices.append(eps)
              else:
                  if log_once("only_full_episodes"):
                      logger.info(
                          "This buffer uses episodes as a storage "
                          "unit and thus allows only full episodes "
                          "to be added to it. Some samples may be "
                          "dropped."
                      )
      elif self.storage_unit == StorageUnit.FRAGMENTS:
          timeslices = [batch]
      else:
          raise ValueError(
              "Unknown `storage_unit={}`".format(self.storage_unit)
          )
  
      for slice in timeslices:
          self.replay_buffers[policy_id].add(slice, **kwargs)
  
imports: import collections
  import logging
  from enum import Enum
  from typing import Any, Dict, Optional
  from ray.util.timer import _Timer
  from ray.rllib.policy.rnn_sequencing import timeslice_along_seq_lens_with_overlap
  from ray.rllib.policy.sample_batch import MultiAgentBatch, SampleBatch
  from ray.rllib.utils.annotations import override
  from ray.rllib.utils.deprecation import Deprecated
  from ray.rllib.utils.from_config import from_config
  from ray.rllib.utils.replay_buffers.replay_buffer import _ALL_POLICIES, ReplayBuffer, StorageUnit
  from ray.rllib.utils.typing import PolicyID, SampleBatchType
  from ray.util.annotations import DeveloperAPI
  from ray.util.debug import log_once
label_desc: Use image node_id 5 for calling the MultiAgentReplayBuffer obj's underlying member method code with example usage: obj._add_to_underlying_buffer(policy_id, batch) without return types