class_docstr: This buffer adds replayed samples to a stream of new experiences.
  
  - Any newly added batch (`add()`) is immediately returned upon
  the next `sample` call (close to on-policy) as well as being moved
  into the buffer.
  - Additionally, a certain number of old samples is mixed into the
  returned sample according to a given "replay ratio".
  - If >1 calls to `add()` are made without any `sample()` calls
  in between, all newly added batches are returned (plus some older samples
  according to the "replay ratio").
  
  .. testcode::
      :skipif: True
  
      # replay ratio 0.66 (2/3 replayed, 1/3 new samples):
      buffer = MultiAgentMixInReplayBuffer(capacity=100,
                                           replay_ratio=0.66)
      buffer.add(<A>)
      buffer.add(<B>)
      buffer.sample(1)
  
  .. testoutput::
  
      ..[<A>, <B>, <B>]
  
  .. testcode::
      :skipif: True
  
      buffer.add(<C>)
      buffer.sample(1)
  
  .. testoutput::
  
      [<C>, <A>, <B>]
      or: [<C>, <A>, <A>], [<C>, <B>, <A>] or [<C>, <B>, <B>],
      but always <C> as it is the newest sample
  
  .. testcode::
      :skipif: True
  
      buffer.add(<D>)
      buffer.sample(1)
  
  .. testoutput::
  
      [<D>, <A>, <C>]
      or [<D>, <A>, <A>], [<D>, <B>, <A>] or [<D>, <B>, <C>], etc..
      but always <D> as it is the newest sample
  
  .. testcode::
      :skipif: True
  
      # replay proportion 0.0 -> replay disabled:
      buffer = MixInReplay(capacity=100, replay_ratio=0.0)
      buffer.add(<A>)
      buffer.sample()
  
  .. testoutput::
  
      [<A>]
  
  .. testcode::
      :skipif: True
  
      buffer.add(<B>)
      buffer.sample()
  
  .. testoutput::
  
      [<B>]
class_docstr_tok: ["This","buffer","adds","replayed","samples","to","a","stream","of","new","experiences",".","-","Any","newly","added","batch","(","`","add","(",")","`",")","is","immediately","returned","upon","the","next","`","sample","`","call","(","close","to","on-policy",")","as","well","as","being","moved","into","the","buffer",".","-","Additionally",",","a","certain","number","of","old","samples","is","mixed","into","the","returned","sample","according","to","a","given","``","replay","ratio","''",".","-","If",">","1","calls","to","`","add","(",")","`","are","made","without","any","`","sample","(",")","`","calls","in","between",",","all","newly","added","batches","are","returned","(","plus","some","older","samples","according","to","the","``","replay","ratio","''",")",".","..","testcode",":",":",":","skipif",":","True","#","replay","ratio","0.66","(","2\/3","replayed",",","1\/3","new","samples",")",":","buffer","=","MultiAgentMixInReplayBuffer","(","capacity=100",",","replay_ratio=0.66",")","buffer.add","(","<","A",">",")","buffer.add","(","<","B",">",")","buffer.sample","(","1",")","..","testoutput",":",":","..","[","<","A",">",",","<","B",">",",","<","B",">","]","..","testcode",":",":",":","skipif",":","True","buffer.add","(","<","C",">",")","buffer.sample","(","1",")","..","testoutput",":",":","[","<","C",">",",","<","A",">",",","<","B",">","]","or",":","[","<","C",">",",","<","A",">",",","<","A",">","]",",","[","<","C",">",",","<","B",">",",","<","A",">","]","or","[","<","C",">",",","<","B",">",",","<","B",">","]",",","but","always","<","C",">","as","it","is","the","newest","sample","..","testcode",":",":",":","skipif",":","True","buffer.add","(","<","D",">",")","buffer.sample","(","1",")","..","testoutput",":",":","[","<","D",">",",","<","A",">",",","<","C",">","]","or","[","<","D",">",",","<","A",">",",","<","A",">","]",",","[","<","D",">",",","<","B",">",",","<","A",">","]","or","[","<","D",">",",","<","B",">",",","<","C",">","]",",","etc","..","but","always","<","D",">","as","it","is","the","newest","sample","..","testcode",":",":",":","skipif",":","True","#","replay","proportion","0.0","-",">","replay","disabled",":","buffer","=","MixInReplay","(","capacity=100",",","replay_ratio=0.0",")","buffer.add","(","<","A",">",")","buffer.sample","(",")","..","testoutput",":",":","[","<","A",">","]","..","testcode",":",":",":","skipif",":","True","buffer.add","(","<","B",">",")","buffer.sample","(",")","..","testoutput",":",":","[","<","B",">","]"]
docstr: Adds a batch to the appropriate policy's replay buffer.
  
  Turns the batch into a MultiAgentBatch of the DEFAULT_POLICY_ID if
  it is not a MultiAgentBatch. Subsequently, adds the individual policy
  batches to the storage.
  
  Args:
      batch: The batch to be added.
      **kwargs: Forward compatibility kwargs.
docstr_tok: ["Adds","a","batch","to","the","appropriate","policy","'s","replay","buffer",".","Turns","the","batch","into","a","MultiAgentBatch","of","the","DEFAULT_POLICY_ID","if","it","is","not","a","MultiAgentBatch",".","Subsequently",",","adds","the","individual","policy","batches","to","the","storage",".","Args",":","batch",":","The","batch","to","be","added",".","*","*","kwargs",":","Forward","compatibility","kwargs","."]
code: def add(self, batch: SampleBatchType, **kwargs) -> None:
      """Adds a batch to the appropriate policy's replay buffer.
  
      Turns the batch into a MultiAgentBatch of the DEFAULT_POLICY_ID if
      it is not a MultiAgentBatch. Subsequently, adds the individual policy
      batches to the storage.
  
      Args:
          batch: The batch to be added.
          **kwargs: Forward compatibility kwargs.
      """
      # Make a copy so the replay buffer doesn't pin plasma memory.
      batch = batch.copy()
      # Handle everything as if multi-agent.
      batch = batch.as_multi_agent()
  
      kwargs = merge_dicts_with_warning(
          self.underlying_buffer_call_args, kwargs
      )
  
      pids_and_batches = self._maybe_split_into_policy_batches(batch)
  
      # We need to split batches into timesteps, sequences or episodes
      # here already to properly keep track of self.last_added_batches
      # underlying buffers should not split up the batch any further
      with self.add_batch_timer:
          if self.storage_unit == StorageUnit.TIMESTEPS:
              for policy_id, sample_batch in pids_and_batches.items():
                  timeslices = sample_batch.timeslices(1)
                  for time_slice in timeslices:
                      self.replay_buffers[policy_id].add(
                          time_slice, **kwargs
                      )
                      self.last_added_batches[policy_id].append(
                          time_slice
                      )
  
          elif self.storage_unit == StorageUnit.SEQUENCES:
              for policy_id, sample_batch in pids_and_batches.items():
                  timeslices = timeslice_along_seq_lens_with_overlap(
                      sample_batch=sample_batch,
                      seq_lens=sample_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,
                  )
                  for slice in timeslices:
                      self.replay_buffers[policy_id].add(
                          slice, **kwargs
                      )
                      self.last_added_batches[policy_id].append(slice)
  
          elif self.storage_unit == StorageUnit.EPISODES:
              for policy_id, sample_batch in pids_and_batches.items():
                  for eps in sample_batch.split_by_episode():
                      # Only add full episodes to the buffer
                      if eps.get(SampleBatch.T)[0] == 0 and (
                          eps.get(SampleBatch.TERMINATEDS, [True])[-1]
                          or eps.get(SampleBatch.TRUNCATEDS, [False])[
                              -1
                          ]
                      ):
                          self.replay_buffers[policy_id].add(
                              eps, **kwargs
                          )
                          self.last_added_batches[policy_id].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:
              for policy_id, sample_batch in pids_and_batches.items():
                  self.replay_buffers[policy_id].add(
                      sample_batch, **kwargs
                  )
                  self.last_added_batches[policy_id].append(
                      sample_batch
                  )
  
      self._num_added += batch.count
  
imports: import collections
  import logging
  import random
  from typing import Any, Dict, Optional
  import numpy
  from ray.rllib.policy.rnn_sequencing import timeslice_along_seq_lens_with_overlap
  from ray.rllib.policy.sample_batch import DEFAULT_POLICY_ID, SampleBatch, concat_samples_into_ma_batch
  from ray.rllib.utils.annotations import override
  from ray.rllib.utils.replay_buffers.multi_agent_prioritized_replay_buffer import MultiAgentPrioritizedReplayBuffer
  from ray.rllib.utils.replay_buffers.multi_agent_replay_buffer import MultiAgentReplayBuffer, ReplayMode, merge_dicts_with_warning
  from ray.rllib.utils.replay_buffers.replay_buffer import _ALL_POLICIES, 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 2 for calling the MultiAgentMixInReplayBuffer obj's underlying member method code with example usage: obj.add(batch) without return types