<jupyter_start><jupyter_text>Implementing Deep Q-Learning with XOR/XNOR data This notebook shows an experiment using rl-keras deep q-learning to classify XOR/XNOR data. It is adapted from <NAME>'s [notebook](https://github.com/NeuroDataDesign/ProgLearn_2021-2022/blob/main/Mordred_Yuan/Week%209/rl_classification.ipynb) testing the performance of deep q-qlearning on the CIFAR 10 dataset.  Setup Installing a Deep Reinforcement Learning for Tensorflow 2 Keras package.<jupyter_code>!pip install -q git+https://github.com/wau/keras-rl2.git<jupyter_output><empty_output><jupyter_text>Importing necessary libraries, as well as the XOR/XNOR generation functions from xor_xnor_functions.py.<jupyter_code>import tensorflow as tf
import tensorflow.keras
from tensorflow.keras.models import Model, Sequential
from keras.layers import (
    Input,
    Dense,
    Conv2D,
    MaxPooling2D,
    Flatten,
    Activation,
    Embedding,
)
from tensorflow.keras.optimizers import Adam, SGD
from tensorflow.keras.layers import LSTM
from keras.preprocessing.sequence import pad_sequences
import numpy as np
import os
import gym
from gym import error, spaces
from gym import utils
from gym.utils import seeding
from sklearn.metrics import classification_report, confusion_matrix
import random
from multiprocessing import Pool
from functools import partial
from keras.layers import *
from keras.models import *
from matplotlib import pyplot as plt
from sklearn.utils import shuffle
from itertools import combinations, product

from proglearn.sims import generate_gaussian_parity
import xor_xnor_functions as fn

import argparse, os
from PIL import Image
import keras.backend as K
from keras.backend import set_session
from rl.agents.dqn import DQNAgent
from rl.policy import LinearAnnealedPolicy, BoltzmannQPolicy, EpsGreedyQPolicy
from rl.memory import SequentialMemory
from rl.core import Processor

from random import sample

os.environ["CUDA_VISIBLE_DEVICES"] = "0"<jupyter_output><empty_output><jupyter_text>Gym Setup We're using Gym to simplify the training. Gym is a toolkit for working with reinforcement algorithms and allows an easy way to test deep q-learning algorithms. We are defining a custom Gym environment for classification here because there is not an included classification environment in Gym. [This](https://towardsdatascience.com/beginners-guide-to-custom-environments-in-openai-s-gym-989371673952) is a helpful article about custom Gym environments and how they are used. In our custom environment, we define init, seed, step, my_metrics, and reset. - Init just defines the parameters the environment uses. Init is pretty straightforward, the most interesting part is the action space. In the XOR/XNOR dataset there are only 2 possible classes, so we set the action space to being 2 discrete classes. In teh CIFAR-10 dataset it would be 10.- Seed is a function that is used if we want to reproduce experiments using random number generators. It's not used in the XOR/XNOR experiment.- Step is the function that takes an action in the q-learning algorithm. This function defines the behavior at each time step and adjusts the q-matrix according to if the sample is correctly identified. We set the reward to 1, and the punishment to -1. It also adds the g-mean and f-measure to metrics at the end of each training run.- My_metrics calculates various metrics for a run. The metrics we are interested in are precision, recall, accuracy, g-mean, and f-measure. In addition, this function prints out these metrics for each run, as well as an overall confusion matrix, g-mean, and f-measure. - Reset simply resets the environment to zero so that it can be used to train again.<jupyter_code>class ClassifyEnv(gym.Env):
    def __init__(self, mode, trainx, trainy):  # mode means training or testing
        self.mode = mode

        self.Env_data = trainx
        # print(type(self.Env_data))
        self.Answer = trainy
        self.id = np.arange(trainx.shape[0])
        # print(self.id)

        self.game_len = self.Env_data.shape[0]

        self.num_classes = 2
        self.action_space = spaces.Discrete(self.num_classes)
        # print(self.action_space)
        self.step_ind = 0
        self.y_pred = []

    def seed(self, seed=None):
        self.np_random, seed = seeding.np_random(seed)
        return [seed]

    def step(self, a):
        self.y_pred.append(a)
        y_true_cur = []
        info = {}
        terminal = False
        if a == self.Answer[self.id[self.step_ind]]:
            reward = 1.0
        else:
            reward = -1.0
            if self.mode == "train":
                terminal = True
        self.step_ind += 1

        if self.step_ind == self.game_len - 1:
            y_true_cur = self.Answer[self.id]
            info["gmean"], info["fmeasure"] = self.My_metrics(
                np.array(self.y_pred), np.array(y_true_cur[: self.step_ind])
            )
            terminal = True

        return self.Env_data[self.id[self.step_ind]], reward, terminal, info

    def My_metrics(self, y_pre, y_true):
        confusion_mat = confusion_matrix(y_true, y_pre)
        print("\n")
        print(classification_report(y_true, y_pre))
        conM = np.array(confusion_mat, dtype="float")
        TP = conM[1][1]
        TN = conM[0][0]
        FN = conM[1][0]
        FP = conM[0][1]
        TPrate = TP / (TP + FN)
        TNrate = TN / (TN + FP)
        FPrate = FP / (TN + FP)
        FNrate = FN / (TP + FN)
        PPvalue = TP / (TP + FP)
        NPvalue = TN / (TN + FN)

        G_mean = np.sqrt(TPrate * TNrate)

        Recall = TPrate = TP / (TP + FN)
        Precision = PPvalue = TP / (TP + FP)
        F_measure = 2 * Recall * Precision / (Recall + Precision)
        print(confusion_mat)
        res = "G-mean:{}, F_measure:{}\n".format(G_mean, F_measure)
        print(res)
        print()
        return G_mean, F_measure

    # return: (states, observations)
    def reset(self):
        if self.mode == "train":
            np.random.shuffle(self.id)
        self.step_ind = 0
        self.y_pred = []
        return self.Env_data[self.id[self.step_ind]]<jupyter_output><empty_output><jupyter_text>Model setup This defines the network that wil be trained.<jupyter_code>def get_image_model(in_shape, output):
    model = Sequential()
    model.add(Dense(500, activation="relu", input_dim=2))
    model.add(Dense(100, activation="relu"))
    model.add(Dense(50, activation="relu"))
    model.add(Dense(2, activation="softmax"))
    return model<jupyter_output><empty_output><jupyter_text>Processor Function<jupyter_code>class ClassifyProcessor(Processor):
    def process_observation(self, observation):
        img = observation.reshape(INPUT_SHAPE)
        processed_observation = np.array(img)
        return processed_observation

    def process_state_batch(self, batch):
        batch = batch.reshape((-1,) + INPUT_SHAPE)
        processed_batch = batch.astype("float32") / 1.0
        return processed_batch

    def process_reward(self, reward):
        return np.clip(reward, -1.0, 1.0)<jupyter_output><empty_output><jupyter_text>XOR/XNOR Setup Get_xor_data and get_xnor_data generate the XOR/XNOR data for the experiment using the functions in xor_xnor_functions.py<jupyter_code>def get_xor_data():
    X_xor, y_xor = generate_gaussian_parity(1000)
    fn.plot_xor_xnor(X_xor, y_xor, "Gaussian XOR")

    return X_xor, y_xor
def get_xnor_data():
    X_xnor, y_xnor = generate_gaussian_parity(1000, angle_params=np.pi / 2)
    fn.plot_xor_xnor(X_xnor, y_xnor, "Gaussian XNOR")

    return X_xnor, y_xnor<jupyter_output><empty_output><jupyter_text>This function is used to divide the XOR/XNOR data into training and testing sets. Input training percentage as a decimal, e.g. .8 for 80% training.<jupyter_code>def divide_train_test(X, Y, train_percentage):
    test_percentage = 1 - train_percentage

    len_train = int(train_percentage * len(X))
    len_test = len(X) - len_train

    full_indices = []
    for i in range(len(X)):
        full_indices.append(i)

    train_indices = set(sample(full_indices, len_train))
    test_indices = set(full_indices) - train_indices

    X_train_subset = []
    for trix in train_indices:
        X_train_subset.append(X[trix])

    X_test_subset = []
    for teix in test_indices:
        X_test_subset.append(X[teix])

    Y_train_subset = []
    for triy in train_indices:
        Y_train_subset.append(Y[triy])

    Y_test_subset = []
    for teiy in test_indices:
        Y_test_subset.append(Y[teiy])

    return (
        np.array(X_train_subset),
        np.array(X_test_subset),
        np.array(Y_train_subset),
        np.array(Y_test_subset),
    )<jupyter_output><empty_output><jupyter_text>Function to run either XOR or XNOR Use this function to train on a set of data. To run XOR, specify 'xor'. To run XNOR, specify 'xnor'. The default is 'xor'. You may also specify the percentage separated into training. The default is a 70-30 training testing split.<jupyter_code>def run_training(x_train, x_test, y_train, y_test):
    # Generate data

    in_shape = x_train.shape[1:]
    num_classes = 2
    mode = "train"
    env = ClassifyEnv(mode, x_train, y_train)
    nb_actions = num_classes
    training_steps = 120000
    model = get_image_model(in_shape, num_classes)

    INPUT_SHAPE = in_shape
    # print(num_classes)
    print(model.summary())

    memory = SequentialMemory(limit=100000, window_length=1)
    processor = ClassifyProcessor()
    policy = LinearAnnealedPolicy(
        EpsGreedyQPolicy(),
        attr="eps",
        value_max=1.0,
        value_min=0.1,
        value_test=0.05,
        nb_steps=100000,
    )
    dqn = DQNAgent(
        model=model,
        nb_actions=2,
        policy=policy,
        memory=memory,
        processor=processor,
        nb_steps_warmup=50000,
        gamma=0.5,
        target_model_update=10000,
        train_interval=4,
        delta_clip=1.0,
    )
    dqn.compile(Adam(learning_rate=0.00025), metrics=["mae"])

    dqn.fit(env, nb_steps=training_steps, log_interval=60000)

    env.mode = "test"
    dqn.test(env, nb_episodes=2, visualize=False)
    env = ClassifyEnv(mode, x_test, y_test)
    env.mode = "test"
    dqn.test(env, nb_episodes=2, visualize=False)<jupyter_output><empty_output><jupyter_text>Train model with just xor and just xnor. It outputs the performance of the model, as well as a confusion matrix.<jupyter_code>x_xor, y_xor = get_xor_data()
x_train_xor, x_test_xor, y_train_xor, y_test_xor = divide_train_test(x_xor, y_xor, 0.7)
INPUT_SHAPE = x_train_xor.shape[1:]
run_training(x_train_xor, x_test_xor, y_train_xor, y_test_xor)<jupyter_output><empty_output><jupyter_text>The XOR data was able to achieve a geometric mean score of .97 and .95 on the first and second test runs respectively.<jupyter_code>x_xnor, y_xnor = get_xnor_data()
x_train_xnor, x_test_xnor, y_train_xnor, y_test_xnor = divide_train_test(
    x_xnor, y_xnor, 0.7
)
INPUT_SHAPE = x_train_xnor.shape[1:]
run_training(x_train_xnor, x_test_xnor, y_train_xnor, y_test_xnor)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.09040873409726609, 'fraction_numerical': 0.014526752684291256, 'mean_word_length': 3.4765751211631666, 'pattern_counts': {'":': 2, '<': 42, '<?xml version=': 0, '>': 42, 'https://': 3, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '10', 'dir': 'jupyter-structured-clean-dedup', 'id': '292410', 'license': 'MIT', 'n_tokens_mistral': 3812, 'n_tokens_neox': 3567, 'n_words': 993, 'path': 'docs/experiments/xor_xnor_with_q_learning.ipynb', 'repo_name': 'levinwil/lifelong_dnns', 'size': '11089'}