Definition. Given a model \(f\), and a data point \(x\), where \(f(x) = y\), we want to find a point \(x'\) where \(f(x') = 1 - y\) such that we minimize \(dist(x, x')\) for any distance function \(dist\).

This definition states that we want to find a point \((x')\) where our binary classifier returns the inverse of the original classification \((y)\).  This definition is also malleable, as we can model different causal relationships of features in order to produce more realistic counterfactual explanations. For example, we can define a set of immutable features which cannot be changed (such as gender and marital status), and a set of non-decreasing features (such as age and education level). We also want to find a counterfactual explanation \((x')\), such that \(x'\) is close to the original training data. Such additional constraints allow us to produce counterfactual explanations that are realistic and well-defined.

1. Correct Output: We need the new data point \(x'\)  to have the desired output from the model (aka moving from a classification of 0 to 1).
2. Minimize Data Manifold Distance: We want to minimize the distance between our new counterfactual data-point \((x')\) and our original training dataset (through KNN distance).
3. Adhere to feature constraints: We have to respect the properties of features especially if they are immutable (cannot be changed) or non-decreasing (must only increase)

• Agent: The agent operates in the entire state space (environment) for a problem. The agent aims to make a right series of actions through the state space to perform some task. It will sample the environment in order to learn the best actions in any state.
• Environment: A generic set of states that produce a reward when an agent moves from state to state. The agent will only learn about the reward from the environment once it has arrived at a certain state.
• Action: At any given state \(s_t\), an agent needs an action to move to state \(s_{t+1}\). The goal of any RL problem is to find a series of actions that move the agent from an undesirable state to a desirable state (by maximizing expected reward through all the states).

Our recent work (appearing soon at AAAI) shows how to use a reinforcement learning framework to generate real-time counterfactual explanations. Let us consider a very simple dataset: \(x_1, x_2, y\). We have some binary classifier that creates a decision boundary shown in Figure 1. Our goal is to move any arbitrary point on the left side of the boundary, namely points in blue, across the boundary to be classified as the opposite class (orange points).

• Agent: In our counterfactual scenario, our agent can be considered to be located on any point on the left side of the decision boundary (any of the blue points). The agent needs to take a series of actions (which we will define below), to move from the left side of the decision boundary to the right side.
• Environment: Our environment (which represents our state-space) is defined as the entire grid (which is discretized into a finite number of states for every input in our feature space). Our environment will send a reward to agents at state \(s\). The reward function is defined as the sum of two different components:
• Classification Reward: We want to reward agents that are on the correct side of the decision boundary. Therefore, we give a high reward once an agent is across the decision boundary. We use the binary classifier’s predict function to determine if an agent is close to the boundary (if the class probability is close to the decision threshold). Therefore the closer to the decision boundary the agent is in, the higher reward it will receive and vice-versa. This value will always be between 0 and 1.
• Dataset Manifold Reward: We want our final counterfactual point to be similar to the training dataset. Therefore, we give higher negative reward for points that are dissimilar to the training data. Computing this similar-ness of states can be implemented in several ways, such as auto-encoders, K-Nearest-Neighbors, etc. For this implementation we used KNN distance (normalized between 0 and 1) to measure the similar-ness of the final counterfactual point.
• Action: In our discretized environment, we allow our agent to move a small distance along any one of the feature dimensions. In our example above, an action \(a\) for an agent at state \(s\), will be a small \((±0.05)\) change for either feature \(x_1\) or \(x_2\). We opted for these small discretized movement to limit to number of movements an agent can perform at any given state. We also believe that an agent will incrementally learn the right movements in order to move towards the decision boundary.

# We create an observation space as a Box (np array) with limits (-1, 1)
self.obeservation_space = gym.spaces.Box(low=np.ones(shape=len(num_features)) * -1,
	                                       high=np.ones(shape=len(num_features)))

# We create an action space as a tuple, where the first value is a feature
# index and the second value is binary (increasing or decreasing)
  self.action_space = gym.spaces.Tuple((
			gym.spaces.Discrete(num_features),
			gym.spaces.Discrete(2)
			))

def step(self, action: Tuple[int, int]) -> Tuple[tuple, float, bool, dict]:
    """
    Step function for the Gym environment

    :param: action: action in the gym environment (transformed_feature_index, increase/decrease)
    :return: state: an observation space defined above
    :return: reward: reward for moving to that state
    :return: done: (bool) is the process complete
    :return: info: dict info about the environment at every step (useful for debugging)
    """
    # Get action and if we should increase or decrease
    feature_index = action[0]
    decrease = bool(action[1])

    # Set default reward to be negative
    reward = -10.0
    done = False
    constant_cost = 0.0

    # Checks to make sure we are not changing an immutable feature
    if feature_index in immutable_features:
        return self.state, reward, done

    # Check to make sure we are not decreasing a non-decreasing features
    if feature_index in non_decreasing_features and decrease:
        return self.state, reward, done

    # Move the agent (X_train should be normalized between 0 and 1)
    new_state = self.state
    amount = -0.05 if decrease else 0.05
    new_state[feature] += amount
    self.state = new_state

    # Compute classifier reward (if we crossed the decision boundary this number
    # will be very large)
    classifier_reward_, done = self.classifier_reward(self.state)

    # Compute KNN reward
    manifold_dist_loss =self.distance_to_data_manifold(self.state)

    # Compute total reward
    reward = classifier_reward - manifold_dist_loss

    return self.state, reward, done, info

def reset(self, initial_state: pd.Series = None) -> tuple:
    """
    Reset methods for Gym environment. Called after an episode is complete or called for starting evaluation with a
    specified starting initial state

    :param: initial_state: initial starting state (used for evaluation)
    """
    if initial_state is not None:
        # This is done for inference
        self.state = initial_state
    else:
        # Randomly get a data point from our training dataset
        self.state = self.X_train.sample()

    return self.state

class FastCFE(gym.Env):
    def __init__(self,
                 classifier_predict: Callable[[pd.Series], np.ndarray],
                 X_train: pd.DataFrame,
                 categorical_features: List[str],
                 immutable_features: List[str] = None,
                 non_decreasing_features: List[str] = None):
        """
        Initializes the Gym Environment to train the RL Model for Counter-Factual-Explanations (CFE)
        (https://arxiv.org/pdf/2106.03962.pdf)
        """
        # Initialize all global class variables
        self.classifier = classifier_predict
        self.X_train = X_train
        self.categorical_features = categorical_features
        self.immutable_features = immutable_features
        self.non_decreasing_features = non_decreasing_features

        # Create the state and action space
        # We create an observation space as a Box (np array) with limits (-1, 1)
        self.obeservation_space = gym.spaces.Box(low=np.ones(shape=len(X_train.columns)) * -1,
                                                 high=np.ones(shape=len(X_train.columns)))


        # We create an action space as a tuple, where the first value is a feature
        # index and the second value is binary (increasing or decreasing)
        self.action_space = gym.spaces.Tuple((gym.spaces.Discrete(len(X_train.columns)),
                            gym.spaces.Discrete(2)))

    def step(self, action: Tuple[int, int]) -> Tuple[tuple, float, bool, dict]:
        """
        Step function for the Gym environment

        :param: action: action in the gym environment (transformed_feature_index, increase/decrease)
        :return: state: an observation space defined above
        :return: reward: reward for moving to that state
        :return: done: (bool) is the process complete
        :return: info: dict info about the environment at every step (useful for debugging)
        """
        # Get action and if we should increase or decrease
        feature_index = action[0]
        decrease = bool(action[1])

        # Set default reward to be negative
        reward = -10.0
        done = False
        constant_cost = 0.0

        # Checks to make sure we are not changing an immutable feature
        if feature_index in immutable_features:
            return self.state, reward, done

        # Check to make sure we are not decreasing a non-decreasing features
        if feature_index in non_decreasing_features and decrease:
            return self.state, reward, done

        # Move the agent (X_train should be normalized between 0 and 1)
        new_state = self.state
        amount = -0.05 if decrease else 0.05
        new_state[feature] += amount
        self.state = new_state

        # Compute classifier reward (if we crossed the decision boundary this number
        # will be very large)
        classifier_reward_, done = self.classifier_reward(self.state)

        # Compute KNN reward
        manifold_dist_loss =self.distance_to_data_manifold(self.state)

        # Compute total reward
        reward = classifier_reward - manifold_dist_loss

        return self.state, reward, done, info

    def reset(self, initial_state: pd.Series = None) -> tuple:
        """
        Reset methods for Gym environment. Called after an episode is complete or called for starting evaluation with a
        specified starting initial state

        :param: initial_state: initial starting state (used for evaluation)
        """
        if initial_state is not None:
            # This is done for inference
            self.state = initial_state
        else:
            # Randomly get a data point from our training dataset
            self.state = self.X_train.sample()

        return self.state

Now that we have created our environment, actions, reward, state-space, and have properly defined our OpenAI Gym environment, we must now produce a model that produces real-time counterfactual explanations. We must produce an optimal policy, \({\pi}(s)\), which will produce an action for an agent in state \(s\) that will maximize future expected reward (reaching our desirable state across the decision boundary). In reinforcement learning, there are several ways to find the optimal policy \({\pi}\), ranging from model-free to model-based optimization techniques. We recommend this link to review more about training reinforcement learning algorithms.

We were focused on finding an optimization algorithm and framework that is fast, scale-able, and easy to use. Much of the recent work has been focused on Distributed Deep Reinforcement Learning, which uses neural networks to implicitly learn the optimal policy \({\pi}\). One such framework is [Ray + Rllib]:

class RayLib():
    def __init__(self, env: FastCFE = None) -> None:
        """
        Initializes a RayLib optimizer for PPO (can add functionally for other optimizers)

        :param: env: the FastCFE class which contains our gym environment
        """
        if env is not None:
            super().__init__(env)

            self.env = env
            self.model: Optional[ppo.PPOTrainer] = None
            self.config: Optional[Dict[str, Any]] = None

    def train(self) -> None:
        """
        Train the RL Agent for the CFE environment (with hyperparameters)

        View more of the features here: (https://docs.ray.io/en/master/rllib-training.html)
        """
        hyperparameters = ppo.DEFAULT_CONFIG.copy()

        # Register the CFE environment to be trained in Ray
        def env_creator(env: FastCFE):
            return env

        register_env("my_env", env_creator)

        # Set the environment config variables
        hyperparameters['env_config'] = self.env

        # Create the PPO Trainer with the configuration and train
        self.model = ppo.PPOTrainer(env="my_env", config=self.config)

        print("Training RL Agent.....")
        for i in range(10):
            result = self.model.train()
            print(f"Training Iteration: {i+1}/{num_epochs}")
            print(f"Total number of datapoints sampled: {result['episodes_total']}")
            print(f"Mean Reward (Max of 100): {result['episode_reward_mean']}")
            print(f"Avg. steps to produce counterfactual (Min: 1, Max: 20): {result['episode_len_mean']}")
            print("_______________________________________________________________________________________")

        return None

    def predict(self, initial_state: pd.Series, max_steps: int = None) -> Tuple[List[pd.Series], float, bool]:
        """
        Make a counterfactual prediction

        :param: initial_state: initial_state
        :param: mode: scaled or unscaled path
        """
        if self.model:
            max_steps = 50
            num = 1
            next_state: Tuple = self.env.reset(initial_state=initial_state)

            done: bool = False
            reward: float = 0.0
            steps = []

            while num < max_steps:
                action = self.model.compute_action(next_state)
                next_state, reward, done, info = self.env.step(action)
                steps.append(next_state)

                if done:
                    break

                num += 1

            return steps, reward, done
        else:
            raise ValueError("Model not loaded")

1. Validity: This is the total number of counterfactual explanations found divided by the total number of data points. This is represented as a percentage.
2. Mean Inference Time: This is the mean time it takes to calculate a batch of inferences. Namely is the time it takes to compute \(n\) inferences divided by \(n\).