Amazon A2I with Amazon SageMaker Model Monitor.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Deriving inference insights for your ML models and sending low confidence predictions to human review workflows using Amazon SageMaker Model Monitor and Amazon A2I\n",
    "\n",
    "Direct marketing, either through mail, email, phone, etc., is a common tactic to acquire customers. Because resources and a customer's attention is limited, the goal is to only target the subset of prospects who are likely to engage with a specific offer. Predicting those potential customers based on readily available information like demographics, past interactions, and environmental factors is a common machine learning problem. In this notebook we present the example problem to predict if a customer will enroll for a term deposit at a bank, after one or more phone calls. We show you how to setup a machine learning workflow on Amazon SageMaker to train a XGBoost Decision Trees algorithm for binary logistic predictions. \n",
    "\n",
    "We will then deploy the model with a real-time endpoint, capture a configurable fraction of the data sent to the endpoint using the SageMaker SDK, create a baseline from the training dataset, launch a monitoring schedule, to let SageMaker Model Monitor inspect collected data and prediction quality, review the generated constraints and statistics, and trigger a human review loop for low confidence predictions. We will then show you how the human loop workers review/update the predictions which generates an annotations dataset. These annotations can be used to update your original training dataset for model re-training.\n",
    "\n",
    "\n",
    "## Contents\n",
    "\n",
    "1. [Preprocess your input dataset](#Preprocess_your_input_dataset)\n",
    "1. [Train and deploy a XGBoost Model](#Step_2_-_Train_and_deploy_a_XGBoost_Model)\n",
    "1. [Generate baselines and start an Amazon SageMaker Model Monitor](#Step_3_-_Start_the_Amazon_SageMaker_Model_Monitor)\n",
    "1. [Review the model monitor reports and derive insights](#Step_4_-_Review_the_model_monitor_reports_and_derive_insights)\n",
    "1. [Setup Human Review loops using Amazon A2I](#Step_5_-_Setup_Human_Review_loops_using_Amazon_A2I)\n",
    "1. [Cleaning up](#Clean_up)\n",
    "1. [Conclusion](#Conclusion)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "## Step 1 - Preprocess your input dataset\n",
    "\n",
    "Let's start by specifying:\n",
    "\n",
    "- The S3 bucket and prefix that you want to use for training and model data.  This should be within the same region as the Notebook Instance, training, and hosting.\n",
    "- The IAM role arn used to give training and hosting access to your data. See the documentation for how to create these.  Note, if more than one role is required for notebook instances, training, and/or hosting, please replace the boto regexp with a the appropriate full IAM role arn string(s)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "isConfigCell": true,
    "tags": [
     "parameters"
    ]
   },
   "outputs": [],
   "source": [
    "bucket = 'a2i-experiments'\n",
    "prefix = 'xgb-dm-mm-a2i-july27'\n",
    " \n",
    "# Define IAM role\n",
    "import boto3\n",
    "import re\n",
    "from sagemaker import get_execution_role\n",
    "\n",
    "role = get_execution_role()\n",
    "print(\"RoleArn: {}\".format(role))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Lets import some data science libraries and the Amazon SageMaker Python SDK"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np                                # For matrix operations and numerical processing\n",
    "import pandas as pd                               # For munging tabular data\n",
    "import matplotlib.pyplot as plt                   # For charts and visualizations\n",
    "from IPython.display import Image                 # For displaying images in the notebook\n",
    "from IPython.display import display               # For displaying outputs in the notebook\n",
    "from time import gmtime, strftime                 # For labeling SageMaker models, endpoints, etc.\n",
    "import sys                                        # For writing outputs to notebook\n",
    "import math                                       # For ceiling function\n",
    "import json                                       # For parsing hosting outputs\n",
    "import os                                         # For manipulating filepath names\n",
    "import sagemaker                                  # Amazon SageMaker's Python SDK provides many helper functions\n",
    "from sagemaker.predictor import csv_serializer    # Converts strings for HTTP POST requests on inference"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Lets download our dataset\n",
    "\n",
    "Let's start by downloading the [direct marketing dataset](https://sagemaker-sample-data-us-west-2.s3-us-west-2.amazonaws.com/autopilot/direct_marketing/bank-additional.zip) from the sample data s3 bucket. \n",
    "\n",
    "\\[Moro et al., 2014\\] S. Moro, P. Cortez and P. Rita. A Data-Driven Approach to Predict the Success of Bank Telemarketing. Decision Support Systems, Elsevier, 62:22-31, June 2014\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!wget https://sagemaker-sample-data-us-west-2.s3-us-west-2.amazonaws.com/autopilot/direct_marketing/bank-additional.zip\n",
    "!apt-get install unzip -y\n",
    "!unzip -o bank-additional.zip"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now lets read this into a Pandas data frame and take a look."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = pd.read_csv('./bank-additional/bank-additional-full.csv')\n",
    "pd.set_option('display.max_columns', 500)     # Make sure we can see all of the columns\n",
    "pd.set_option('display.max_rows', 10)         # Keep the output on one page\n",
    "data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's talk about the data.  At a high level, we can see:\n",
    "\n",
    "* We have a little over 40K customer records, and 20 features for each customer\n",
    "* The features are mixed; some numeric, some categorical\n",
    "* The data appears to be sorted, at least by `time` and `contact`, maybe more\n",
    "* The target variable we want to predict is Column \"Y\": Has the client subscribed a term deposit? (binary: 'yes','no')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, let's look at how our features relate to the target that we are attempting to predict."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "for column in data.select_dtypes(include=['object']).columns:\n",
    "    print(column)\n",
    "    if column != 'y':\n",
    "        display(pd.crosstab(index=data[column], columns=data['y'], normalize='columns'))\n",
    "\n",
    "for column in data.select_dtypes(exclude=['object']).columns:\n",
    "    print(column)\n",
    "    hist = data[[column, 'y']].hist(by='y', bins=30)\n",
    "    plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Notice that:\n",
    "\n",
    "* Customers who are-- \"blue-collar\", \"married\", \"unknown\" default status, contacted by \"telephone\", and/or in \"may\" are a substantially lower portion of \"yes\" than \"no\" for subscribing.\n",
    "* Distributions for numeric variables are different across \"yes\" and \"no\" subscribing groups, but the relationships may not be straightforward or obvious."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Data cleanup and transformation\n",
    "\n",
    "Our dataset requires some simple transformations for the algorithm to understand features. Also XGBoost handles missing values by default so we perform a few conversions as below"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data['no_previous_contact'] = np.where(data['pdays'] == 999, 1, 0)                                 # Indicator variable to capture when pdays takes a value of 999\n",
    "data['not_working'] = np.where(np.in1d(data['job'], ['student', 'retired', 'unemployed']), 1, 0)   # Indicator for individuals not actively employed\n",
    "model_data = pd.get_dummies(data)                                                                  # Convert categorical variables to sets of indicators"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's remove the economic features and `duration` from our data as they would need to be forecasted with high precision to use as inputs in future predictions. Even if we were to use values of the economic indicators from the previous quarter, this value is likely not as relevant for prospects contacted early in the next quarter as those contacted later on."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model_data = model_data.drop(['duration',\n",
    "                              'emp.var.rate',\n",
    "                              'cons.price.idx',\n",
    "                              'cons.conf.idx',\n",
    "                              'euribor3m',\n",
    "                              'nr.employed'],\n",
    "                             axis=1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Lets convert our input data into training, validation and test datasets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Randomly sort the data then split out first 70%, second 20%, and last 10%\n",
    "train_data, validation_data, test_data = np.split(model_data.sample(frac=1, random_state=1729), [int(0.7 * len(model_data)), int(0.9 * len(model_data))])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_data.head(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(train_data.shape)\n",
    "print(validation_data.shape)\n",
    "print(test_data.shape)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Some other versions of the training dataset to experiment with for training/baselining"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_data.drop(['y_no', 'y_yes'], axis=1).to_csv(path_or_buf='train_no_y.csv', index=False, header=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_data.drop(['y_no', 'y_yes'], axis=1).to_csv(path_or_buf='train_no_y_with_header.csv', index=False, header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_data.head(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_data.shape"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Amazon SageMaker's XGBoost container expects data in the libSVM or CSV data format.  For this example, we'll stick to CSV.  Note that the first column must be the target variable and the CSV should not include headers.  Also, notice that although repetitive it's easiest to do this after the train|validation|test split rather than before.  This avoids any misalignment issues due to random reordering."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_df_with_header = pd.concat([train_data['y_yes'],\n",
    "           train_data.drop(['y_no', 'y_yes'], axis=1)],\n",
    "          axis=1\n",
    "         )\n",
    "valid_df_with_header = pd.concat([validation_data['y_yes'],\n",
    "           validation_data.drop(['y_no', 'y_yes'], axis=1)],\n",
    "          axis=1\n",
    "         )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "train_df_with_header.to_csv('train.csv', index=False, header=False)\n",
    "train_df_with_header.to_csv('train_with_header.csv', index=False, header=True)\n",
    "valid_df_with_header.to_csv('validation.csv', index=False, header=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we'll copy the file to S3 for Amazon SageMaker's managed training to pickup."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# If training/validating with other files, make sure to upload them to s3 here.\n",
    "boto3.Session().resource('s3').Bucket(bucket).Object(os.path.join(prefix, 'train/train.csv')\n",
    "                                                    ).upload_file('train.csv')\n",
    "boto3.Session().resource('s3').Bucket(bucket).Object(os.path.join(prefix, 'validation/validation.csv')\n",
    "                                                    ).upload_file('validation.csv')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 2 - Train and deploy a XGBoost model\n",
    "\n",
    "The `XGBoost` (eXtreme Gradient Boosting) is a popular and efficient open-source implementation of the gradient boosted trees algorithm. Gradient boosting is a supervised learning algorithm that attempts to accurately predict a target variable by combining an ensemble of estimates from a set of simpler and weaker models. The XGBoost algorithm performs well in machine learning competitions because of its robust handling of a variety of data types, relationships, distributions, and the variety of hyperparameters that you can fine-tune. You can use XGBoost for regression, classification (binary and multiclass), and ranking problems.\n",
    "\n",
    "You can use the new release of the XGBoost algorithm either as an Amazon SageMaker built-in algorithm or as a framework to run training scripts in your local environments. Using the built-in algorithm version of XGBoost is simpler than using the open source version, because you don’t have to write a training script. If you don’t need the features and flexibility of open source XGBoost, consider using the built-in version. For information about using the Amazon SageMaker XGBoost built-in algorithm, see [XGBoost Algorithm](https://docs.aws.amazon.com/sagemaker/latest/dg/xgboost.html) in the Amazon SageMaker Developer Guide.\n",
    "\n",
    "First we'll need to specify the ECR container location for Amazon SageMaker's implementation of XGBoost."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.amazon.amazon_estimator import get_image_uri\n",
    "container = get_image_uri(boto3.Session().region_name, 'xgboost', '1.0-1')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Then, because we're training with the CSV file format, we'll create `s3_input`s that our training function can use as a pointer to the files in S3, which also specifies that the content type is CSV."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "s3_input_train = sagemaker.s3_input(s3_data='s3://{}/{}/train'.format(bucket, prefix),\n",
    "                                    content_type='csv')\n",
    "s3_input_validation = sagemaker.s3_input(s3_data='s3://{}/{}/validation/'.format(bucket, prefix),\n",
    "                                         content_type='csv')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create the XGBoost Estimator\n",
    "\n",
    "Now that we have the XGBoost container, we use it to construct an estimator using the SageMaker Estimator API and initiate a training job. This XGBoost built-in algorithm runs directly on the input datasets. Amazon SageMaker XGBoost currently only trains using CPUs. It is a memory-bound (as opposed to compute-bound) algorithm. So, a general-purpose compute instance (for example, M5) is a better choice than a compute-optimized instance (for example, C4). Further, we recommend that you have enough total memory in selected instances to hold the training data. Although it supports the use of disk space to handle data that does not fit into main memory (the out-of-core feature available with the libsvm input mode), writing cache files onto disk slows the algorithm processing time.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sess = sagemaker.Session()\n",
    "\n",
    "xgb = sagemaker.estimator.Estimator(container,\n",
    "                                    role, \n",
    "                                    train_instance_count=1, \n",
    "                                    train_instance_type='ml.m5.2xlarge',\n",
    "                                    output_path='s3://{}/{}/output'.format(bucket, prefix),\n",
    "                                    sagemaker_session=sess)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Specify Hyperparameters\n",
    "\n",
    "Hyperparameters are set by users to facilitate the estimation of model parameters from data. \n",
    "\n",
    "- `max_depth` Controls how deep each tree within the algorithm can be built. Deeper trees can lead to better fit, but are more computationally expensive and can lead to overfitting. Typically, you need to explore some trade-offs in model performance between a large number of shallow trees and a smaller number of deeper trees.\n",
    "- `subsample` Controls sampling of the training data. This hyperparameter can help reduce overfitting, but setting it too low can also starve the model of data.\n",
    "- `num_round` Controls the number of boosting rounds. This value specifies the models that are subsequently trained using the residuals of previous iterations. Again, more rounds should produce a better fit on the training data, but can be computationally expensive or lead to overfitting.\n",
    "- `eta Controls` how aggressive each round of boosting is. Larger values lead to more conservative boosting.\n",
    "- `gamma` Controls how aggressively trees are grown. Larger values lead to more conservative models.\n",
    "- `min_child_weight` Also controls how aggresively trees are grown. Large values lead to a more conservative model.\n",
    "\n",
    "For other hyperparameters and to know more details please refer to [XGBoost Parameters](https://xgboost.readthedocs.io/en/release_0.90/parameter.html#parameters-for-tree-booster) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "xgb.set_hyperparameters(max_depth=5,\n",
    "                        eta=0.2,\n",
    "                        gamma=4,\n",
    "                        min_child_weight=6,\n",
    "                        subsample=0.8,\n",
    "                        silent=0,\n",
    "                        objective='binary:logistic',\n",
    "                        num_round=100)\n",
    "\n",
    "xgb.fit({'train': s3_input_train, 'validation': s3_input_validation}) "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Deploy the XGBoost Model\n",
    "\n",
    "Now that we've trained the `XGBoost` algorithm on our data, let's deploy a model that's hosted behind a real-time endpoint. As a first step lets specify the paths to Amazon S3 locations for storing data, report, and processing code."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(f\"Bucket is: {bucket}\")\n",
    "print(f\"Prefix is: {prefix}\")\n",
    "data_capture_prefix = '{}/datacapture'.format(prefix)\n",
    "s3_capture_upload_path = 's3://{}/{}'.format(bucket, data_capture_prefix)\n",
    "reports_prefix = '{}/reports'.format(prefix)\n",
    "s3_report_path = 's3://{}/{}'.format(bucket,reports_prefix)\n",
    "code_prefix = '{}/code'.format(prefix)\n",
    "#s3_code_preprocessor_uri = 's3://{}/{}/{}'.format(bucket,code_prefix, 'preprocessor.py')\n",
    "#s3_code_postprocessor_uri = 's3://{}/{}/{}'.format(bucket,code_prefix, 'postprocessor.py')\n",
    "\n",
    "print(\"Capture path: {}\".format(s3_capture_upload_path))\n",
    "print(\"Report path: {}\".format(s3_report_path))\n",
    "#print(\"Preproc Code path: {}\".format(s3_code_preprocessor_uri))\n",
    "#print(\"Postproc Code path: {}\".format(s3_code_postprocessor_uri))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We now specify the capture option called [DataCaptureConfig](https://docs.aws.amazon.com/sagemaker/latest/APIReference/API_DataCaptureConfig.html). You can capture the request payload, the response payload, or both with this configuration. The capture configuration applies to all variants."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.model_monitor import DataCaptureConfig\n",
    "\n",
    "endpoint_name = 'DEMO-xgb-churn-pred-model-monitor-' + strftime(\"%Y-%m-%d-%H-%M-%S\", gmtime())\n",
    "print(\"EndpointName={}\".format(endpoint_name))\n",
    "\n",
    "data_capture_config = DataCaptureConfig(\n",
    "                        enable_capture=True,\n",
    "                        sampling_percentage=100,\n",
    "                        destination_s3_uri=s3_capture_upload_path)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "After you fit an XGBoost Estimator, you can host the newly created model in SageMaker. After you call fit, you can call deploy on an XGBoost estimator to create a SageMaker endpoint. The endpoint runs a SageMaker-provided XGBoost model server to host your model, which was run when you called fit. \n",
    "\n",
    "The deploy function returns a Predictor object, which you can use to do inference on the Endpoint hosting your XGBoost model. Each Predictor provides a predict method which can do inference with numpy arrays, Python lists, or strings. After inference arrays or lists are serialized and sent to the XGBoost model server, predict returns the result of inference against your model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "xgb_predictor = xgb.deploy(initial_instance_count=1,\n",
    "                           instance_type='ml.m5.2xlarge',\n",
    "                           endpoint_name=endpoint_name,\n",
    "                           data_capture_config=data_capture_config)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "xgb_predictor.content_type = 'text/csv'\n",
    "xgb_predictor.serializer = csv_serializer"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Invoke the deployed model using the endpoint\n",
    "\n",
    "You can now send data to this endpoint to get inferences in real time. Because you enabled the data capture in the previous steps, the request and response payload, along with some additional metadata, is saved in the Amazon S3 location that you specified in DataCaptureConfig."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Step 32\n",
    "# Use this code to instantiate the predictor object if you've already created an earlier endpoint. Uncomment both code lines below\n",
    "# endpoint_name = \"DEMO-xgb-churn-pred-model-monitor-2020-07-22-21-33-23\"\n",
    "#xgb_predictor_2 = sagemaker.predictor.RealTimePredictor(endpoint=endpoint_name,content_type='text/csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "def invoke_endpoint(predictor, data_df, rows=500):\n",
    "    print(\"Sending test traffic to the endpoint {}. \\nPlease wait...\".format(endpoint_name))\n",
    "    \n",
    "    predictions = ''\n",
    "    i = 0\n",
    "    for row in data_df.to_numpy():\n",
    "        payload = \",\".join([str(num) for num in row])\n",
    "#         print(array)\n",
    "        response = predictor.predict(payload)\n",
    "        if i % 10 == 0:\n",
    "            print(response)\n",
    "        predictions = ','.join([predictions, response.decode('utf-8')])\n",
    "#         time.sleep(0.5)\n",
    "        i = i + 1\n",
    "\n",
    "#     print(predictions)\n",
    "    return np.fromstring(predictions[1:], sep=',')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "payload_df = test_data.drop(['y_no', 'y_yes'], axis=1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "payload_df.head(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# if this is not your first time running this notebook and you already have an endpoint running, \n",
    "# please execute the cell marked Step 32 above following the instructions provided in comments and\n",
    "# replace the xgb_predictor variable name below accordingly\n",
    "predictions = invoke_endpoint(xgb_predictor, payload_df.head(200))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The Data Capture Config we specified earlier for the endpoint should have reported the input sent to the endpoint, the outout received from the endpoint as well as some metrics. Lets verify if the report was sent to S3 "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "s3_client = boto3.Session().client('s3')\n",
    "current_endpoint_capture_prefix = '{}/{}'.format(data_capture_prefix, endpoint_name)\n",
    "result = s3_client.list_objects(Bucket=bucket, Prefix=current_endpoint_capture_prefix)\n",
    "capture_files = [capture_file.get(\"Key\") for capture_file in result.get('Contents')]\n",
    "print(\"Found Capture Files:\")\n",
    "print(\"\\n \".join(capture_files))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Lets now review the content of the S3 objects "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "\n",
    "def get_obj_body(obj_key):\n",
    "    return s3_client.get_object(Bucket=bucket, Key=obj_key).get('Body').read().decode(\"utf-8\")\n",
    "\n",
    "capture_file = get_obj_body(capture_files[-1])\n",
    "\n",
    "print(json.dumps(json.loads(capture_file.split('\\n')[0]), indent=2))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Understanding the prediction result\n",
    "\n",
    "The objective we defined in the Hyperparameters for model training was `binary:logistic`. The model will apply logistic regression for binary classification, with output as probability. The probability refers to the log likelihood of the bernoulli distribution. For more details refer to [Bernoulli Distribution](https://en.wikipedia.org/wiki/Bernoulli_distribution). In our example above the value of 0.035 in the output indicates a 4% probability that the customer will accept the offer. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 3 - Start the Amazon SageMaker Model Monitor\n",
    "\n",
    "Amazon SageMaker Model Monitor continuously monitors the quality of Amazon SageMaker machine learning models in production. It enables developers to set alerts for when there are deviations in the model quality, such as data drift. Early and pro-active detection of these deviations enables you to take corrective actions, such as retraining models, auditing upstream systems, or fixing data quality issues without having to monitor models manually or build additional tooling. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create a Baseline\n",
    "\n",
    "The baseline calculations of statistics and constraints are needed as a standard against which data drift and other data quality issues can be detected. Amazon SageMaker Model Monitor provides a built-in container that provides the ability to suggest the constraints automatically for CSV and flat JSON input. This sagemaker-model-monitor-analyzer container also provides you with a range of model monitoring capabilities, including constraint validation against a baseline, and emitting Amazon CloudWatch metrics. This container is based on Spark and is built with Deequ.\n",
    "\n",
    "The training dataset that you used to trained the model is usually a good baseline dataset. The training dataset data schema and the inference dataset schema should exactly match (the number and order of the features). Note that the prediction/output column(s) are assumed to be the 1st column(s) in the training dataset. From the training dataset, you can ask Amazon SageMaker to suggest a set of baseline constraints and generate descriptive statistics to explore the data. For this example, upload the training dataset that was used to train the pretrained model included in this example. If you already have it in Amazon S3, you can point to it directly."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# copy over the training dataset to Amazon S3 (if you already have it in Amazon S3, you could reuse it)\n",
    "baseline_prefix = prefix + '/baselining'\n",
    "baseline_data_prefix = baseline_prefix + '/data'\n",
    "baseline_results_prefix = baseline_prefix + '/results'\n",
    "\n",
    "baseline_data_uri = 's3://{}/{}'.format(bucket,baseline_data_prefix)\n",
    "baseline_results_uri = 's3://{}/{}'.format(bucket, baseline_results_prefix)\n",
    "print('Baseline data uri: {}'.format(baseline_data_uri))\n",
    "print('Baseline results uri: {}'.format(baseline_results_uri))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Upload the training file for baselining\n",
    "training_data_file = open(\"train.csv\", 'rb')\n",
    "s3_key = os.path.join(baseline_prefix, 'data', 'train.csv')\n",
    "boto3.Session().resource('s3').Bucket(bucket).Object(s3_key).upload_fileobj(training_data_file)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Start the baseline job\n",
    "from sagemaker.model_monitor import DefaultModelMonitor\n",
    "from sagemaker.model_monitor.dataset_format import DatasetFormat\n",
    "\n",
    "my_default_monitor = DefaultModelMonitor(\n",
    "    role=role,\n",
    "    instance_count=1,\n",
    "    instance_type='ml.m5.4xlarge',\n",
    "    volume_size_in_gb=100,\n",
    "    max_runtime_in_seconds=3600,\n",
    ")\n",
    "\n",
    "my_default_monitor.suggest_baseline(\n",
    "    baseline_dataset=baseline_data_uri+'/train.csv',\n",
    "    dataset_format=DatasetFormat.csv(header=False), # changed this to header=False since train.csv does not have header. \n",
    "    output_s3_uri=baseline_results_uri,\n",
    "    wait=True\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Inspect baseline job results\n",
    "\n",
    "Now the baseline job has completed, lets inspect the results. Two files are generated:\n",
    "- `statistics.json` This file is expected to have columnar statistics for each feature in the dataset that is analyzed. See the schema for this file in the [Schema for Statistics](https://docs.aws.amazon.com/sagemaker/latest/dg/model-monitor-byoc-statistics.html).\n",
    "- `constraints.json` This file is expected to have the constraints on the features observed. See the schema for this file in the [Schema for Constraints](https://docs.aws.amazon.com/sagemaker/latest/dg/model-monitor-byoc-constraints.html)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "s3_client = boto3.Session().client('s3')\n",
    "result = s3_client.list_objects(Bucket=bucket, Prefix=baseline_results_prefix)\n",
    "report_files = [report_file.get(\"Key\") for report_file in result.get('Contents')]\n",
    "print(\"Found Files:\")\n",
    "print(\"\\n \".join(report_files))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Lets inspect the contents of the statstics.json file for a couple of entries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# if header was set to False for the baselining creation function, the column names will look like \"_c0,\" \"_c1,\" etc.\n",
    "import pandas as pd\n",
    "\n",
    "baseline_job = my_default_monitor.latest_baselining_job\n",
    "schema_df = pd.json_normalize(baseline_job.baseline_statistics().body_dict[\"features\"])\n",
    "schema_df.head(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Lets inspect the contents of the constraints.json file for a couple of entries. This should contain constraints applied for each of the columns. In our case we see that the non-negative constraint is applied"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "constraints_df = pd.json_normalize(baseline_job.suggested_constraints().body_dict[\"features\"])\n",
    "constraints_df.head(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Lets check the monitoring configurations specified\n",
    "constraints_mon_df = pd.json_normalize(baseline_job.suggested_constraints().body_dict[\"monitoring_config\"])\n",
    "constraints_mon_df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The Monitoring Configuration determines the Monitor's actions. For more details please refer to [Model Monitor documentation](https://docs.aws.amazon.com/sagemaker/latest/dg/model-monitor.html)\n",
    "\n",
    "- `emit_metrics` Amazon SageMaker emits Cloudwatch metrics for each feature/column observed in the dataset in the /aws/sagemaker/Endpoints/data-metric namespace with EndpointName and ScheduleName dimensions\n",
    "- `datatype_check_threshold` During the baseline step, the generated constraints suggest the inferred data type for each column. The monitoring_config.datatype_check_threshold parameter can be tuned to adjust the threshold on when it is flagged as a violation\n",
    "- `domain_content_threshold` If there are more unknown values for a String field in the current dataset than in the baseline dataset, this threshold can be used to dictate if it needs to be flagged as a violation\n",
    "- `distribution_constraints.comparison_threshold` **This value is used to calculate model drift.** If the threshold is above the value set for the comparison_threshold, this causes a failure that is treated as a violation in the violation report. The distance is calculated by getting the maximum absolute difference between the cumulative distribution functions of two distributions.\n",
    "\n",
    "For more details on constraints please refer to [Constraints Schema](https://docs.aws.amazon.com/sagemaker/latest/dg/model-monitor-byoc-constraints.html)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create Monitoring Schedule\n",
    "\n",
    "With a Monitoring Schedule, Amazon SageMaker can kick off processing jobs at a specified frequency to analyze the data collected during a given period. Amazon SageMaker provides a pre-built container for performing analysis on tabular datasets. In the processing job, Amazon SageMaker compares the dataset for the current analysis with the baseline statistics, constraints provided and generate a violations report. In addition, CloudWatch metrics are emitted for each feature under analysis. Lets create a monitoring schedule to run hourly"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sagemaker.model_monitor import CronExpressionGenerator\n",
    "from time import gmtime, strftime\n",
    "\n",
    "mon_schedule_name = 'xgb-mm-dm-a2i-blog-monitor-schedule-' + strftime(\"%Y-%m-%d-%H-%M-%S\", gmtime())\n",
    "my_default_monitor.create_monitoring_schedule(\n",
    "    monitor_schedule_name=mon_schedule_name,\n",
    "    endpoint_input=xgb_predictor.endpoint,\n",
    "    output_s3_uri=s3_report_path,\n",
    "    statistics=my_default_monitor.baseline_statistics(),\n",
    "    constraints=my_default_monitor.suggested_constraints(),\n",
    "    schedule_cron_expression=CronExpressionGenerator.hourly(),\n",
    "    enable_cloudwatch_metrics=True,\n",
    "\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Note:** Please ensure that the MaxRunTime for your Model Monitor is smaller than the CRON schedule you specify. Otherwise you will get an error - CreateMonitoringSchedule operation: stopping condition should be smaller than schedule cadence"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Lets invoke the endpoint continuously to generate traffic for the model monitor to pickup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from threading import Thread\n",
    "from time import sleep\n",
    "import time\n",
    "\n",
    "endpoint_name=xgb_predictor.endpoint\n",
    "runtime_client = boto3.client('runtime.sagemaker')\n",
    "\n",
    "# (just repeating code from above for convenience/ able to run this section independently)\n",
    "def invoke_endpoint(ep_name, file_name, runtime_client):\n",
    "    with open(file_name, 'r') as f:\n",
    "        for row in f:\n",
    "            payload = row.rstrip('\\n')\n",
    "#             print(payload)\n",
    "            response = runtime_client.invoke_endpoint(EndpointName=ep_name,\n",
    "                                          ContentType='text/csv', \n",
    "                                          Body=payload)\n",
    "            response['Body'].read()\n",
    "            time.sleep(1)\n",
    "            \n",
    "def invoke_endpoint_forever():\n",
    "    while True:\n",
    "        print(\"invoke\")\n",
    "        invoke_endpoint(endpoint_name, 'train_no_y.csv', runtime_client)\n",
    "        \n",
    "thread = Thread(target = invoke_endpoint_forever)\n",
    "thread.start()\n",
    "\n",
    "# Note that you need to stop the kernel to stop the invocations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Check the Monitor Schedule status\n",
    "desc_schedule_result = my_default_monitor.describe_schedule()\n",
    "print('Schedule status: {}'.format(desc_schedule_result['MonitoringScheduleStatus']))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 4 - Review Model Monitoring Execution Output"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Check executation status every 15 minutes\n",
    "mon_executions = my_default_monitor.list_executions()\n",
    "print(\"We created a hourly schedule above and it will kick off executions ON the hour (plus 0 - 20 min buffer.\\nWe will check execution status every 15 minutes...\")\n",
    "\n",
    "while len(mon_executions) == 0:\n",
    "    print(\"Waiting for the 1st execution to happen...\")\n",
    "    time.sleep(900)\n",
    "    mon_executions = my_default_monitor.list_executions()    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Lets look at the latest execution status and print the report name"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "latest_execution = mon_executions[-1] # latest execution's index is -1, second to last is -2 and so on..\n",
    "time.sleep(60)\n",
    "latest_execution.wait(logs=False)\n",
    "\n",
    "print(\"Latest execution status: {}\".format(latest_execution.describe()['ProcessingJobStatus']))\n",
    "print(\"Latest execution result: {}\".format(latest_execution.describe()['ExitMessage']))\n",
    "\n",
    "latest_job = latest_execution.describe()\n",
    "if (latest_job['ProcessingJobStatus'] != 'Completed'):\n",
    "        print(\"====STOP==== \\n No completed executions to inspect further. Please wait till an execution completes or investigate previously reported failures.\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "report_uri=latest_execution.output.destination\n",
    "print('Report Uri: {}'.format(report_uri))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Review Violations Report\n",
    "\n",
    "The violations file is generated as the output of a MonitoringExecution, which lists the results of evaluating the constraints (specified in the constraints.json file) against the current dataset that was analyzed. The Amazon SageMaker Model Monitor pre-built container provides the following violation checks:\n",
    "\n",
    "- `data_type_check` \n",
    "- `completeness_check`\n",
    "- `baseline_drift_check`\n",
    "- `missing_column_check`\n",
    "- `extra_column_check`\n",
    "- `categorical_values_check`\n",
    "\n",
    "For more details about the violation checks please refer to the Model Monitor documentation [here](https://docs.aws.amazon.com/sagemaker/latest/dg/model-monitor-interpreting-violations.html)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "violations = my_default_monitor.latest_monitoring_constraint_violations()\n",
    "constraints_df = pd.json_normalize(violations.body_dict[\"violations\"])\n",
    "constraints_df.head(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As can be seen above, the model monitor detected a data type mismatch violation in one of the requests sent to the endpoint. To enable proactive action on these metrics, please check the documentation [here](https://docs.aws.amazon.com/sagemaker/latest/dg/model-monitor-interpreting-cloudwatch.html) for how to emit these metrics to Amazon Cloudwatch. You can also visualize the results of monitoring in Amazon SageMaker Studio. For information about the onboarding process for using Studio, see [Onboard to Amazon SageMaker Studio](https://docs.aws.amazon.com/sagemaker/latest/dg/gs-studio-onboard.html). You can view monitoring results at your endpoints using charts. You can view the jobs being monitored and deep dive into each of the jobs. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# View distribution of output probabilities\n",
    "from matplotlib import pyplot as plt\n",
    "plt.xlim([-0.1, 1.1])\n",
    "bin_size=0.05\n",
    "bins = np.arange(-0.1, 1.1, bin_size) # fixed bin size\n",
    "\n",
    "plt.hist(predictions, bins=bins, alpha=0.5)\n",
    "plt.title('Distribution of probabilities')\n",
    "plt.xlabel(f'probabilities (bin size = {bin_size})')\n",
    "plt.ylabel('count')\n",
    "\n",
    "plt.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Evaluate results\n",
    "\n",
    "There are two perspectives to be considered here to determine next steps for our experiment. \n",
    "\n",
    "- **`Model Monitor Violations`** We only saw the datatype_check violation from the Model Monitor. We did not see a model drift violation. This indicates that the test dataset we used for endpoint invocations, did not result in an absolute difference of > 10% of distribution (comparison_threshold is set to 0.1 in the constraints.json) to the baseline dataset. So the prediction results, even if they tend to be lower probabilities for offer acceptance are aligned with the results in the training dataset \n",
    "\n",
    "- **`Probability Distribution of Prediction results`** From the prediction results we captured earlier and the outputs, we see that the model tends to distribute probabilities for most of the customers in the 0 to 30% range for likelihood of accepting an offer based on a sample of 200 requests we sent from our test dataset. If we re-run the inference for all requests in the test dataset we will see that the distributions are aligned in the 0 to 30% range and > 60% range. And since requests in this range will not trigger a model drift violation, these results are as expected for the model. But if we see consistent results with a prediction output probability of > 30% and < 60% we need to review these results as over time they will trigger a model drift. \n",
    "\n",
    "So as a next step let's setup a human loop to review the prediction results we received with a > 30% and < 60% probability output"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 5 - Set up a human review loop for low-confidence detection using Amazon A2I"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Amazon Augmented AI (Amazon A2I) makes it easy to build the workflows required for human review of ML predictions. Amazon A2I brings human review to all developers, removing the undifferentiated heavy lifting associated with building human review systems or managing large numbers of human reviewers.\n",
    "\n",
    "To incorporate Amazon A2I into your human review workflows, you need three resources:\n",
    "\n",
    "A worker task template to create a worker UI. The worker UI displays your input data, such as documents or images, and instructions to workers. It also provides interactive tools that the worker uses to complete your tasks. For more information, see A2I instructions overview\n",
    "\n",
    "A human review workflow, also referred to as a flow definition. You use the flow definition to configure your human workforce and provide information about how to accomplish the human review task. For built-in task types, you also use the flow definition to identify the conditions under which a review human loop is triggered. For example, with Amazon Rekognition can perform image content moderation using machine learning. You can use the flow definition to specify that an image will be sent to a human for content moderation review if Amazon Rekognition's confidence score output is low for your use case. You can create a flow definition in the Amazon Augmented AI console or with the Amazon A2I APIs. To learn more about both of these options, see create flow definition\n",
    "\n",
    "A human loop to start your human review workflow. When you use one of the built-in task types, the corresponding AWS service creates and starts a human loop on your behalf when the conditions specified in your flow definition are met or for each object if no conditions were specified. When a human loop is triggered, human review tasks are sent to the workers as specified in the flow definition.\n",
    "\n",
    "When using a custom task type, you start a human loop using the Amazon Augmented AI Runtime API. When you call StartHumanLoop in your custom application, a task is sent to human reviewers."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### In this section, you set up a human review loop for low-confidence detections in Amazon A2I. It includes the following steps:\n",
    "\n",
    "* Create or choose your workforce\n",
    "* Create a human task UI\n",
    "* Create the flow definition\n",
    "* Trigger conditions for human loop activation\n",
    "* Check the human loop status and wait for reviewers to complete the task"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create your workforce\n",
    "\n",
    "This step requires you to use the AWS Console. We will create a private workteam and add only one user (you) to it. To create a private team:\n",
    "\n",
    "1. Go to AWS Console > Amazon SageMaker > Labeling workforces\n",
    "1. Click \"Private\" and then \"Create private team\".\n",
    "1. Enter the desired name for your private workteam.\n",
    "1. Enter your own email address in the \"Email addresses\" section.\n",
    "1. Enter the name of your organization and a contact email to administer the private workteam.\n",
    "1. Click \"Create Private Team\".\n",
    "1. The AWS Console should now return to AWS Console > Amazon SageMaker > Labeling workforces. Your newly created team should be visible under \"Private teams\". Next to it you will see an ARN which is a long string that looks like arn:aws:sagemaker:region-name-123456:workteam/private-crowd/team-name. **Please enter this ARN in the cell below**\n",
    "1. You should get an email from no-reply@verificationemail.com that contains your workforce username and password.\n",
    "1. In AWS Console > Amazon SageMaker > Labeling workforces, click on the URL in Labeling portal sign-in URL. Use the email/password combination from Step 8 to log in (you will be asked to create a new, non-default password).\n",
    "1. This is your private worker's interface. When we create a verification task in Verify your task using a private team below, your task should appear in this window. You can invite your colleagues to participate in the labeling job by clicking the \"Invite new workers\" button."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "WORKTEAM_ARN= \"<your workteam ARN created in Step 7 above>\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Lets now initialize some variables that we need in the subsequent steps"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import io\n",
    "import uuid\n",
    "import time\n",
    "\n",
    "timestamp = time.strftime(\"%Y-%m-%d-%H-%M-%S\", time.gmtime())\n",
    "# Amazon SageMaker client\n",
    "sagemaker_client = boto3.client('sagemaker', region)\n",
    "\n",
    "# Amazon Augment AI (A2I) client\n",
    "a2i = boto3.client('sagemaker-a2i-runtime')\n",
    "\n",
    "# Amazon S3 client \n",
    "s3 = boto3.client('s3', region)\n",
    "\n",
    "# Flow definition name - this value is unique per account and region. You can also provide your own value here.\n",
    "flowDefinitionName = 'fd-sagemaker-mm-a2i-tabular-' + timestamp\n",
    "\n",
    "# Task UI name - this value is unique per account and region. You can also provide your own value here.\n",
    "taskUIName = 'ui-sagemaker-mm-a2i-tabular-' + timestamp"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create the human task UI\n",
    "\n",
    "Create a human task UI resource, giving a UI template in liquid html. This template will be rendered to the human workers whenever human loop is required. For over 70 pre built UIs, check: https://github.com/aws-samples/amazon-a2i-sample-task-uis"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "template = r\"\"\"\n",
    "<script src=\"https://assets.crowd.aws/crowd-html-elements.js\"></script>\n",
    "\n",
    "<style>\n",
    "  table, tr, th, td {\n",
    "    border: 1px solid black;\n",
    "    border-collapse: collapse;\n",
    "    padding: 5px;\n",
    "  }\n",
    "</style>\n",
    "\n",
    "<crowd-form>\n",
    "    <div>\n",
    "        <h1>Instructions</h1>\n",
    "        <p>Please review the predictions in the Predictions table based on the input data table below, and make corrections where appropriate. </p>\n",
    "        <p> Here are the labels: </p>\n",
    "        <p> 0: Customer enrolled (i.e., made a deposit) </p>\n",
    "        <p> 1: Customer did NOT enroll (i.e., did NOT make a deposit) </p>\n",
    "    </div>\n",
    "    <div>\n",
    "      <h3> Bank direct marketing dataset </h3>\n",
    "      <div id=\"my_table\"> {{ task.input.table | skip_autoescape }} </div>\n",
    "   </div>\n",
    "    <br>\n",
    "    <h1> Predictions Table </h1>\n",
    "    <p>\n",
    "        Note that probabilities of less than 0.5 indicate a prediction of 0 \n",
    "        and probabilities of greater than or equal to 0.5 indicate a prediction of 1.\n",
    "        The column \"AGREE/DISAGREE WITH ML PREDICTION\" refers to whether or not you agree \n",
    "        with the prediction based on the probability.\n",
    "    </p>\n",
    "    <table>\n",
    "      <tr>\n",
    "        <th>ROW NUMBER</th>\n",
    "        <th>MODEL PREDICTION  (Probability of enrolling)</th>\n",
    "        <th>AGREE/DISAGREE WITH PREDICTION?</th>\n",
    "        <th>YOUR PREDICTION (0 or 1)</th>\n",
    "        <th>CHANGE REASON </th>\n",
    "      </tr>\n",
    "\n",
    "      {% for pair in task.input.Pairs %}\n",
    "\n",
    "        <tr>\n",
    "          <td>{{ pair.row }}</td>\n",
    "          <td><crowd-text-area name=\"predicted{{ forloop.index }}\" value=\"{{ pair.prediction }}\"></crowd-text-area></td>\n",
    "          <td>\n",
    "            <p>\n",
    "              <input type=\"radio\" id=\"agree{{ forloop.index }}\" name=\"rating{{ forloop.index }}\" value=\"agree\" required>\n",
    "              <label for=\"agree{{ forloop.index }}\">Agree</label>\n",
    "            </p>\n",
    "            <p>\n",
    "              <input type=\"radio\" id=\"disagree{{ forloop.index }}\" name=\"rating{{ forloop.index }}\" value=\"disagree\" required>\n",
    "              <label for=\"disagree{{ forloop.index }}\">Disagree</label>       \n",
    "            </p> \n",
    "          </td>\n",
    "          <td>\n",
    "            <p>\n",
    "            <input type=\"text\" name=\"True Prediction {{ forloop.index }}\" placeholder=\"Enter your Prediction\" />\n",
    "            </p>\n",
    "           </td>\n",
    "           <td>\n",
    "            <p>\n",
    "            <input type=\"text\" name=\"Change Reason {{ forloop.index }}\" placeholder=\"Explain why you changed the prediction\" />\n",
    "            </p>\n",
    "           </td>\n",
    "        </tr>\n",
    "\n",
    "      {% endfor %}\n",
    "\n",
    "    </table>\n",
    "</crowd-form>\n",
    "\"\"\"\n",
    "\n",
    "def create_task_ui():\n",
    "    '''\n",
    "    Creates a Human Task UI resource.\n",
    "\n",
    "    Returns:\n",
    "    struct: HumanTaskUiArn\n",
    "    '''\n",
    "    response = sagemaker_client.create_human_task_ui(\n",
    "        HumanTaskUiName=taskUIName,\n",
    "        UiTemplate={'Content': template})\n",
    "    return response"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create task UI\n",
    "humanTaskUiResponse = create_task_ui()\n",
    "humanTaskUiArn = humanTaskUiResponse['HumanTaskUiArn']\n",
    "print(humanTaskUiArn)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create the Flow Definition\n",
    "In this section, we're going to create a flow definition definition. Flow Definitions allow us to specify:\n",
    "- The workforce that your tasks will be sent to. \n",
    "- The instructions that your workforce will receive. This is called a worker task template. \n",
    "- Where your output data will be stored.\n",
    "\n",
    "This demo is going to use the API, but you can optionally create this workflow definition in the console as well. For more details and instructions, see: https://docs.aws.amazon.com/sagemaker/latest/dg/a2i-create-flow-definition.html."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "create_workflow_definition_response = sagemaker_client.create_flow_definition(\n",
    "        FlowDefinitionName= flowDefinitionName,\n",
    "        RoleArn= role,\n",
    "        HumanLoopConfig= {\n",
    "            \"WorkteamArn\": WORKTEAM_ARN,\n",
    "            \"HumanTaskUiArn\": humanTaskUiArn,\n",
    "            \"TaskCount\": 1,\n",
    "            \"TaskDescription\": \"Make sure the labels are correct\",\n",
    "            \"TaskTitle\": \"tabular data a2i demo\"\n",
    "        },\n",
    "        OutputConfig={\n",
    "            \"S3OutputPath\" : OUTPUT_PATH\n",
    "        }\n",
    "    )\n",
    "flowDefinitionArn = create_workflow_definition_response['FlowDefinitionArn'] # let's save this ARN for future use"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Describe flow definition - status should be active\n",
    "for x in range(60):\n",
    "    describeFlowDefinitionResponse = sagemaker_client.describe_flow_definition(FlowDefinitionName=flowDefinitionName)\n",
    "    print(describeFlowDefinitionResponse['FlowDefinitionStatus'])\n",
    "    if (describeFlowDefinitionResponse['FlowDefinitionStatus'] == 'Active'):\n",
    "        print(\"Flow Definition is active\")\n",
    "        break\n",
    "    time.sleep(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Set trigger conditions for human loop activation\n",
    "\n",
    "As we discussed before, we see from the probability distribution of predicted results that a prediction output probability range of > 30% and < 60% when inferred consistently may lead to model drift and needs to be investigated. So we setup the trigger condition for the Amazon A2I human loop to be within this range. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# assign our original dataset \n",
    "model_data_categorical = model_data  \n",
    "\n",
    "LOWER_THRESHOLD = 0.3\n",
    "UPPER_THRESHOLD = 0.6\n",
    "small_payload_df = model_data_categorical.head(len(predictions))\n",
    "small_payload_df['prediction_prob'] = predictions\n",
    "small_payload_df = small_payload_df.loc[\n",
    "    (small_payload_df['prediction_prob'] > LOWER_THRESHOLD) &\n",
    "    (small_payload_df['prediction_prob'] < UPPER_THRESHOLD)\n",
    "]\n",
    "print(small_payload_df.shape)\n",
    "small_payload_df.head(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(f\"{len(small_payload_df)} out of {len(predictions)} samples or \" +\n",
    "      '{:.1%} of the payload was sent to review.'.format(len(small_payload_df)/len(predictions)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Note that the prediction is in terms of a probability from 0 to 1\n",
    "low_conf_predictions = small_payload_df['prediction_prob'].to_list()\n",
    "NUM_TO_REVIEW = len(low_conf_predictions) # You can change this number as desired\n",
    "item_list = [{'row': \"ROW_{}\".format(x), 'prediction': low_conf_predictions[x]} for x in range(NUM_TO_REVIEW)]\n",
    "item_list"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ip_content = {\"table\": small_payload_df.reset_index().drop(columns = ['index']).head(NUM_TO_REVIEW).to_html(), \n",
    "              'Pairs': item_list\n",
    "             }"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Activate human loops\n",
    "import json\n",
    "humanLoopName = str(uuid.uuid4())\n",
    "\n",
    "start_loop_response = a2i.start_human_loop(\n",
    "            HumanLoopName=humanLoopName,\n",
    "            FlowDefinitionArn=flowDefinitionArn,\n",
    "            HumanLoopInput={\n",
    "                \"InputContent\": json.dumps(ip_content)\n",
    "            }\n",
    "        )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Check status of task completion and human loop\n",
    "\n",
    "Let's define a function that allows us to check the status of Human Loop progress"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "completed_human_loops = []\n",
    "resp = a2i.describe_human_loop(HumanLoopName=humanLoopName)\n",
    "print(f'HumanLoop Name: {humanLoopName}')\n",
    "print(f'HumanLoop Status: {resp[\"HumanLoopStatus\"]}')\n",
    "print(f'HumanLoop Output Destination: {resp[\"HumanLoopOutput\"]}')\n",
    "print('\\n')\n",
    "    \n",
    "if resp[\"HumanLoopStatus\"] == \"Completed\":\n",
    "    completed_human_loops.append(resp)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Wait for workers to complete their tasks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "workteamName = WORKTEAM_ARN[WORKTEAM_ARN.rfind('/') + 1:]\n",
    "print(\"Navigate to the private worker portal and do the tasks. Make sure you've invited yourself to your workteam!\")\n",
    "print('https://' + sagemaker_client.describe_workteam(WorkteamName=workteamName)['Workteam']['SubDomain'])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Check status of human loop again"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "completed_human_loops = []\n",
    "resp = a2i.describe_human_loop(HumanLoopName=humanLoopName)\n",
    "print(f'HumanLoop Name: {humanLoopName}')\n",
    "print(f'HumanLoop Status: {resp[\"HumanLoopStatus\"]}')\n",
    "print(f'HumanLoop Output Destination: {resp[\"HumanLoopOutput\"]}')\n",
    "print('\\n')\n",
    "    \n",
    "if resp[\"HumanLoopStatus\"] == \"Completed\":\n",
    "    completed_human_loops.append(resp)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's inspect the results of the human review tasks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import re\n",
    "import pprint\n",
    "\n",
    "pp = pprint.PrettyPrinter(indent=4)\n",
    "\n",
    "for resp in completed_human_loops:\n",
    "    splitted_string = re.split('s3://' +  bucket + '/', resp['HumanLoopOutput']['OutputS3Uri'])\n",
    "    output_bucket_key = splitted_string[1]\n",
    "\n",
    "    response = s3.get_object(Bucket=bucket, Key=output_bucket_key)\n",
    "    content = response[\"Body\"].read()\n",
    "    json_output = json.loads(content)\n",
    "    pp.pprint(json_output)\n",
    "    print('\\n')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Clean up\n",
    "\n",
    "If you are done with this notebook, please run the cell below.  This will remove the hosted endpoint you created. Also please make sure to stop this notebook instance when you are done to avoid incurring charges"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sagemaker.Session().delete_endpoint(xgb_predictor.endpoint)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Conclusion\n",
    "\n",
    "This notebook demonstrated how you can use Amazon SageMaker Model Monitor and Amazon A2I to setup a monitoring schedule for your Amazon SageMaker model endpoints, specify baselines that include constraint thresholds, observe inference traffic, derive insights such as model drift, completeness, data type violations and send the low confidence predictions to a Human Workflow with labelers to review and update the results. The human labeled output can be used to augment the training dataset for re-training and improving model accuracy."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "conda_python3",
   "language": "python",
   "name": "conda_python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.10"
  },
  "notice": "Copyright 2017 Amazon.com, Inc. or its affiliates. All Rights Reserved.  Licensed under the Apache License, Version 2.0 (the \"License\"). You may not use this file except in compliance with the License. A copy of the License is located at http://aws.amazon.com/apache2.0/ or in the \"license\" file accompanying this file. This file is distributed on an \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License."
 },
 "nbformat": 4,
 "nbformat_minor": 4
}