class_docstr: Counterfactual Reinforcement Learning. 
class_docstr_tok: ["Counterfactual","Reinforcement","Learning","."]
docstr: Fit the model agnostic counterfactual generator.
  
  Parameters
  ----------
  X
      Training data array.
  
  Returns
  -------
  self
      The explainer itself.
code: def fit(self, X: np.ndarray) -> "Explainer":
      """
      Fit the model agnostic counterfactual generator.
  
      Parameters
      ----------
      X
          Training data array.
  
      Returns
      -------
      self
          The explainer itself.
      """
      # Define boolean flag for initializing actor and critic network for Tensorflow backend.
      initialize_actor_critic = False
  
      # Define replay buffer (this will deal only with numpy arrays).
      replay_buff = ReplayBuffer(size=self.params["replay_buffer_size"])
  
      # Define noise variable.
      noise = NormalActionNoise(mu=0, sigma=self.params["act_noise"])
  
      # Define data generator.
      data_generator = self.backend.data_generator(X=X, **self.params)
      data_iter = iter(data_generator)
  
      for step in tqdm(range(self.params["train_steps"])):
          # Sample training data.
          try:
              data = next(data_iter)
          except StopIteration:
              if hasattr(data_generator, "on_epoch_end"):
                  # This is just for tensorflow backend.
                  data_generator.on_epoch_end()
  
              data_iter = iter(data_generator)
              data = next(data_iter)
  
          # Add None condition if condition does not exist.
          if "C" not in data:
              data["C"] = None
  
          # Compute input embedding.
          Z = self.backend.encode(X=data["X"], **self.params)
          data.update({"Z": Z})
  
          # Compute counterfactual embedding.
          Z_cf = self.backend.generate_cf(**data, **self.params)
          data.update({"Z_cf": Z_cf})
  
          # Add noise to the counterfactual embedding.
          Z_cf_tilde = self.backend.add_noise(
              noise=noise, step=step, **data, **self.params
          )
          data.update({"Z_cf_tilde": Z_cf_tilde})
  
          # Decode noised counterfactual and apply postprocessing step to X_cf_tilde.
          X_cf_tilde = self.backend.decode(
              Z=data["Z_cf_tilde"], **self.params
          )
  
          for pp_func in self.params["postprocessing_funcs"]:
              # Post-process noised counterfactual.
              X_cf_tilde = pp_func(
                  self.backend.to_numpy(X_cf_tilde),
                  self.backend.to_numpy(data["X"]),
                  self.backend.to_numpy(data["C"]),
              )
          data.update({"X_cf_tilde": X_cf_tilde})
  
          # Compute model's prediction on the noised counterfactual
          X_cf_tilde = self.params["decoder_inv_preprocessor"](
              self.backend.to_numpy(data["X_cf_tilde"])
          )
          Y_m_cf_tilde = self.params["predictor"](X_cf_tilde)
  
          # Compute reward.
          R_tilde = self.params["reward_func"](
              self.backend.to_numpy(Y_m_cf_tilde),
              self.backend.to_numpy(data["Y_t"]),
          )
          data.update(
              {"R_tilde": R_tilde, "Y_m_cf_tilde": Y_m_cf_tilde}
          )
  
          # Store experience in the replay buffer.
          data = {
              key: self.backend.to_numpy(data[key])
              for key in data.keys()
          }
          replay_buff.append(**data)
  
          if (
              step % self.params["update_every"] == 0
              and step > self.params["update_after"]
          ):
              for i in range(self.params["update_every"]):
                  # Sample batch of experience form the replay buffer.
                  sample = replay_buff.sample()
  
                  # Initialize actor and critic. This is required for tensorflow in order to reinitialize the
                  # explainer object and call fit multiple times. If the models are not reinitialized, the
                  # error: "tf.function-decorated function tried to create variables on non-first call" is raised.
                  # This is due to @tf.function and building the model for the first time in a compiled function
                  if (
                      not initialize_actor_critic
                      and self.params["backend"] == Framework.TENSORFLOW
                  ):
                      self.backend.initialize_actor_critic(
                          **sample, **self.params
                      )
                      self.backend.initialize_optimizers(
                          **sample, **self.params
                      )
                      initialize_actor_critic = True
  
                  if "C" not in sample:
                      sample["C"] = None
  
                  # Decode counterfactual. This procedure has to be done here and not in the experience loop
                  # since the actor is updating but old experience is used. Thus, the decoding of the counterfactual
                  # will not correspond to the latest actor network. Remember that the counterfactual is used
                  # for the consistency loss. The counterfactual generation is performed here due to @tf.function
                  # which does not allow all post-processing functions.
                  Z_cf = self.backend.generate_cf(
                      Z=self.backend.to_tensor(
                          sample["Z"], **self.params
                      ),
                      Y_m=self.backend.to_tensor(
                          sample["Y_m"], **self.params
                      ),
                      Y_t=self.backend.to_tensor(
                          sample["Y_t"], **self.params
                      ),
                      C=self.backend.to_tensor(
                          sample["C"], **self.params
                      ),
                      **self.params,
                  )
  
                  X_cf = self.backend.decode(Z=Z_cf, **self.params)
                  for pp_func in self.params["postprocessing_funcs"]:
                      # Post-process counterfactual.
                      X_cf = pp_func(
                          self.backend.to_numpy(X_cf),
                          self.backend.to_numpy(sample["X"]),
                          self.backend.to_numpy(sample["C"]),
                      )
  
                  # Add counterfactual instance to the sample to be used in the update function for consistency loss
                  sample.update(
                      {
                          "Z_cf": self.backend.to_numpy(Z_cf),
                          "X_cf": self.backend.to_numpy(X_cf),
                      }
                  )
  
                  # Update critic by one-step gradient descent.
                  losses = self.backend.update_actor_critic(
                      **sample, **self.params
                  )
  
                  # Convert all losses from tensors to numpy arrays.
                  losses = {
                      key: self.backend.to_numpy(losses[key]).item()
                      for key in losses.keys()
                  }
  
                  # Call all callbacks.
                  for callback in self.params["callbacks"]:
                      callback(
                          step=step,
                          update=i,
                          model=self,
                          sample=sample,
                          losses=losses,
                      )
      return self
  
imports: import logging
  import os
  from abc import ABC, abstractmethod
  from copy import deepcopy
  from typing import TYPE_CHECKING, Any, Callable, Dict, Optional, Tuple, Union, cast
  import numpy
  from tqdm import tqdm
  from alibi.api.defaults import DEFAULT_DATA_CFRL, DEFAULT_META_CFRL
  from alibi.api.interfaces import Explainer, Explanation, FitMixin
  from alibi.explainers.backends.cfrl_base import generate_empty_condition, get_classification_reward, get_hard_distribution, identity_function
  from alibi.utils.frameworks import Framework, has_pytorch, has_tensorflow
label_desc: Use image node_id 10 for calling the CounterfactualRL obj's underlying member method code with example usage: obj.fit(X) and returns: self