code: def update_priorities_in_replay_buffer(
      replay_buffer: ReplayBuffer,
      config: AlgorithmConfigDict,
      train_batch: SampleBatchType,
      train_results: ResultDict,
  ) -> None:
      """Updates the priorities in a prioritized replay buffer, given training results.
  
      The `abs(TD-error)` from the loss (inside `train_results`) is used as new
      priorities for the row-indices that were sampled for the train batch.
  
      Don't do anything if the given buffer does not support prioritized replay.
  
      Args:
          replay_buffer: The replay buffer, whose priority values to update. This may also
              be a buffer that does not support priorities.
          config: The Algorithm's config dict.
          train_batch: The batch used for the training update.
          train_results: A train results dict, generated by e.g. the `train_one_step()`
              utility.
      """
      # Only update priorities if buffer supports them.
      if isinstance(replay_buffer, MultiAgentPrioritizedReplayBuffer):
          # Go through training results for the different policies (maybe multi-agent).
          prio_dict = {}
          for policy_id, info in train_results.items():
              # TODO(sven): This is currently structured differently for
              #  torch/tf. Clean up these results/info dicts across
              #  policies (note: fixing this in torch_policy.py will
              #  break e.g. DDPPO!).
              td_error = info.get(
                  "td_error", info[LEARNER_STATS_KEY].get("td_error")
              )
  
              policy_batch = train_batch.policy_batches[policy_id]
              # Set the get_interceptor to None in order to be able to access the numpy
              # arrays directly (instead of e.g. a torch array).
              policy_batch.set_get_interceptor(None)
              # Get the replay buffer row indices that make up the `train_batch`.
              batch_indices = policy_batch.get("batch_indexes")
  
              if SampleBatch.SEQ_LENS in policy_batch:
                  # Batch_indices are represented per column, in order to update
                  # priorities, we need one index per td_error
                  _batch_indices = []
  
                  # Sequenced batches have been zero padded to max_seq_len.
                  # Depending on how batches are split during learning, not all
                  # sequences have an associated td_error (trailing ones missing).
                  if policy_batch.zero_padded:
                      seq_lens = len(td_error) * [
                          policy_batch.max_seq_len
                      ]
                  else:
                      seq_lens = policy_batch[SampleBatch.SEQ_LENS][
                          : len(td_error)
                      ]
  
                  # Go through all indices by sequence that they represent and shrink
                  # them to one index per sequences
                  sequence_sum = 0
                  for seq_len in seq_lens:
                      _batch_indices.append(batch_indices[sequence_sum])
                      sequence_sum += seq_len
                  batch_indices = np.array(_batch_indices)
  
              if td_error is None:
                  if log_once(
                      "no_td_error_in_train_results_from_policy_{}".format(
                          policy_id
                      )
                  ):
                      logger.warning(
                          "Trying to update priorities for policy with id `{}` in "
                          "prioritized replay buffer without providing td_errors in "
                          "train_results. Priority update for this policy is being "
                          "skipped.".format(policy_id)
                      )
                  continue
  
              if batch_indices is None:
                  if log_once(
                      "no_batch_indices_in_train_result_for_policy_{}".format(
                          policy_id
                      )
                  ):
                      logger.warning(
                          "Trying to update priorities for policy with id `{}` in "
                          "prioritized replay buffer without providing batch_indices in "
                          "train_batch. Priority update for this policy is being "
                          "skipped.".format(policy_id)
                      )
                  continue
  
              #  Try to transform batch_indices to td_error dimensions
              if len(batch_indices) != len(td_error):
                  T = replay_buffer.replay_sequence_length
                  assert (
                      len(batch_indices) > len(td_error)
                      and len(batch_indices) % T == 0
                  )
                  batch_indices = batch_indices.reshape([-1, T])[:, 0]
                  assert len(batch_indices) == len(td_error)
              prio_dict[policy_id] = (batch_indices, td_error)
  
          # Make the actual buffer API call to update the priority weights on all
          # policies.
          replay_buffer.update_priorities(prio_dict)
  
file_path: ray/rllib/utils/replay_buffers/utils.py
imports: import logging
  import psutil
  from typing import Optional, Any
  import numpy
  from ray.rllib.utils import deprecation_warning
  from ray.rllib.utils.annotations import DeveloperAPI
  from ray.rllib.utils.deprecation import DEPRECATED_VALUE
  from ray.rllib.utils.from_config import from_config
  from ray.rllib.utils.metrics.learner_info import LEARNER_STATS_KEY
  from ray.rllib.utils.replay_buffers import MultiAgentPrioritizedReplayBuffer, ReplayBuffer, MultiAgentReplayBuffer
  from ray.rllib.policy.sample_batch import concat_samples, MultiAgentBatch, SampleBatch
  from ray.rllib.utils.typing import ResultDict, SampleBatchType, AlgorithmConfigDict
  from ray.util import log_once
label_desc: Use image node_id 1 for calling a global function with example usage: update_priorities_in_replay_buffer(replay_buffer, config, train_batch, train_results) without return types