class_docstr: GSSPlugin replay plugin.
  
  Code adapted from the repository:
  https://github.com/RaptorMai/online-continual-learning
  Handles an external memory fulled with samples selected
  using the Greedy approach of GSS algorithm.
  `before_forward` callback is used to process the current
  sample and estimate a score.
docstr: After every forward this function select sample to fill
  the memory buffer based on cosine similarity
docstr_tok: ["After","every","forward","this","function","select","sample","to","fill","the","memory","buffer","based","on","cosine","similarity"]
code: def after_forward(
      self, strategy, num_workers=0, shuffle=True, **kwargs
  ):
      """
      After every forward this function select sample to fill
      the memory buffer based on cosine similarity
      """
  
      strategy.model.eval()
  
      # Compute the gradient dimension
      grad_dims = []
      for param in strategy.model.parameters():
          grad_dims.append(param.data.numel())
  
      place_left = (
          self.ext_mem_list_x.size(0) - self.ext_mem_list_current_index
      )
      if place_left <= 0:  # buffer full
          batch_sim, mem_grads = self.get_batch_sim(
              strategy,
              grad_dims,
              batch_x=strategy.mb_x,
              batch_y=strategy.mb_y,
          )
  
          if batch_sim < 0:
              buffer_score = self.buffer_score[
                  : self.ext_mem_list_current_index
              ].cpu()
  
              buffer_sim = (buffer_score - torch.min(buffer_score)) / (
                  (torch.max(buffer_score) - torch.min(buffer_score))
                  + 0.01
              )
  
              # draw candidates for replacement from the buffer
              index = torch.multinomial(
                  buffer_sim, strategy.mb_x.size(0), replacement=False
              ).to(strategy.device)
  
              # estimate the similarity of each sample in the received batch
              # to the randomly drawn samples from the buffer.
              batch_item_sim = self.get_each_batch_sample_sim(
                  strategy,
                  grad_dims,
                  mem_grads,
                  strategy.mb_x,
                  strategy.mb_y,
              )
  
              # normalize to [0,1]
              scaled_batch_item_sim = (
                  (batch_item_sim + 1) / 2
              ).unsqueeze(1)
              buffer_repl_batch_sim = (
                  (self.buffer_score[index] + 1) / 2
              ).unsqueeze(1)
              # draw an event to decide on replacement decision
              outcome = torch.multinomial(
                  torch.cat(
                      (scaled_batch_item_sim, buffer_repl_batch_sim),
                      dim=1,
                  ),
                  1,
                  replacement=False,
              )
              # replace samples with outcome =1
              added_indx = torch.arange(
                  end=batch_item_sim.size(0), device=strategy.device
              )
              sub_index = outcome.squeeze(1).bool()
              self.ext_mem_list_x[index[sub_index]] = strategy.mb_x[
                  added_indx[sub_index]
              ].clone()
              self.ext_mem_list_y[index[sub_index]] = strategy.mb_y[
                  added_indx[sub_index]
              ].clone()
              self.buffer_score[index[sub_index]] = batch_item_sim[
                  added_indx[sub_index]
              ].clone()
      else:
          offset = min(place_left, strategy.mb_x.size(0))
          updated_mb_x = strategy.mb_x[:offset]
          updated_mb_y = strategy.mb_y[:offset]
  
          # first buffer insertion
          if self.ext_mem_list_current_index == 0:
              batch_sample_memory_cos = (
                  torch.zeros(updated_mb_x.size(0)) + 0.1
              )
          else:
              # draw random samples from buffer
              mem_grads = self.get_rand_mem_grads(
                  strategy=strategy,
                  grad_dims=grad_dims,
                  gss_batch_size=len(strategy.mb_x),
              )
  
              # estimate a score for each added sample
              batch_sample_memory_cos = self.get_each_batch_sample_sim(
                  strategy,
                  grad_dims,
                  mem_grads,
                  updated_mb_x,
                  updated_mb_y,
              )
  
          curr_idx = self.ext_mem_list_current_index
          self.ext_mem_list_x[curr_idx : curr_idx + offset].data.copy_(
              updated_mb_x
          )
          self.ext_mem_list_y[curr_idx : curr_idx + offset].data.copy_(
              updated_mb_y
          )
          self.buffer_score[curr_idx : curr_idx + offset].data.copy_(
              batch_sample_memory_cos
          )
          self.ext_mem_list_current_index += offset
  
      strategy.model.train()
  
file_path: avalanche/avalanche/training/plugins/gss_greedy.py
imports: from typing import TYPE_CHECKING
  import torch
  from avalanche.benchmarks.utils import _make_taskaware_classification_dataset
  from avalanche.benchmarks.utils.data_loader import ReplayDataLoader
  from avalanche.training.plugins.strategy_plugin import SupervisedPlugin
label_desc: Use image node_id 9 for calling the GSS_greedyPlugin obj's underlying member method code with example usage: obj.after_forward(strategy, num_workers, shuffle) without return types