insurance_claims_pipeline.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Insurance claims demo with SCv2\n",
    "\n",
    "This notebook implements a simple insurance claims processing pipeline, while showcase a number of SCv2 features."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "First off, make sure you have Seldon Core v2 up and running. For instructions please refer to the README.me found in this repo.\n",
    "\n",
    "We will be using Seldon CLI a lot, let's check it."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "/home/andrei/projects/seldon-core/operator/bin/seldon\r\n"
     ]
    }
   ],
   "source": [
    "!which seldon"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This notebook can be used all models running on Triton runtime, or with a mix of some models running on Triton and others on MLServer. To make a choice, use the switch variable below."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "use_triton_only = False"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Stage 1 - basic pipeline\n",
    "\n",
    "In the first part of the demo we will implement the basic pipeline as displayed here\n",
    "\n",
    "![insurance claims pipeline](https://raw.githubusercontent.com/mlatcl/fbp-vs-soa/main/insurance_claims/diagrams/insurance_claims_fbp_min.png?raw=true)\n",
    "\n",
    "As you can see, the pipeline takes in a claim, classifies it (in several steps) as simple and complex, and generates corresponding payout to the claimant. All circles here represent business logic nodes, while rectangles are data streams that hold input, output and interim data. Because of its dataflow stream-base nature, SCv2 is an ideal tool to build such data processing pipelines."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We have implemented each step of the pipeline as a custom Python code wrapped as a Triton model. Their source code and specs can be inspected in the [models](./models) folder in this repo. For example, here are their yaml definitions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "apiVersion: mlops.seldon.io/v1alpha1\n",
      "kind: Model\n",
      "metadata:\n",
      "  name: calculate_claim_value\n",
      "  namespace: seldon-mesh\n",
      "spec:\n",
      "  storageUri: \"/mnt/models/insurance_claims/calculate_claim_value\"\n",
      "  requirements:\n",
      "  - triton\n",
      "  - python\n",
      "----------------------------------------------------------------------\n",
      "apiVersion: mlops.seldon.io/v1alpha1\n",
      "kind: Model\n",
      "metadata:\n",
      "  name: classify_claim_value\n",
      "  namespace: seldon-mesh\n",
      "spec:\n",
      "  storageUri: \"/mnt/models/insurance_claims/classify_claim_value\"\n",
      "  requirements:\n",
      "  - triton\n",
      "  - python\n",
      "----------------------------------------------------------------------\n",
      "apiVersion: mlops.seldon.io/v1alpha1\n",
      "kind: Model\n",
      "metadata:\n",
      "  name: classify_claim_complexity\n",
      "  namespace: seldon-mesh\n",
      "spec:\n",
      "  storageUri: \"/mnt/models/insurance_claims/classify_claim_complexity_mlserver\"\n",
      "  requirements:\n",
      "  - mlserver\n",
      "----------------------------------------------------------------------\n",
      "apiVersion: mlops.seldon.io/v1alpha1\n",
      "kind: Model\n",
      "metadata:\n",
      "  name: calculate_complex_claim_payout\n",
      "  namespace: seldon-mesh\n",
      "spec:\n",
      "  storageUri: \"/mnt/models/insurance_claims/calculate_complex_claim_payout_mlserver\"\n",
      "  requirements:\n",
      "  - mlserver\n",
      "----------------------------------------------------------------------\n",
      "apiVersion: mlops.seldon.io/v1alpha1\n",
      "kind: Model\n",
      "metadata:\n",
      "  name: calculate_simple_claim_payout\n",
      "  namespace: seldon-mesh\n",
      "spec:\n",
      "  storageUri: \"/mnt/models/insurance_claims/calculate_simple_claim_payout_mlserver\"\n",
      "  requirements:\n",
      "  - mlserver\n"
     ]
    }
   ],
   "source": [
    "!cat ./models/insurance_claims/calculate_claim_value.yaml\n",
    "print(\"----------------------------------------------------------------------\")\n",
    "!cat ./models/insurance_claims/classify_claim_value.yaml\n",
    "print(\"----------------------------------------------------------------------\")\n",
    "\n",
    "if use_triton_only:\n",
    "    !cat ./models/insurance_claims/classify_claim_complexity.yaml\n",
    "    print(\"----------------------------------------------------------------------\")\n",
    "    !cat ./models/insurance_claims/calculate_complex_claim_payout.yaml\n",
    "    print(\"----------------------------------------------------------------------\")\n",
    "    !cat ./models/insurance_claims/calculate_simple_claim_payout.yaml\n",
    "else:\n",
    "    !cat ./models/insurance_claims/classify_claim_complexity_mlserver.yaml\n",
    "    print(\"----------------------------------------------------------------------\")\n",
    "    !cat ./models/insurance_claims/calculate_complex_claim_payout_mlserver.yaml\n",
    "    print(\"----------------------------------------------------------------------\")\n",
    "    !cat ./models/insurance_claims/calculate_simple_claim_payout_mlserver.yaml\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Notice how in the yaml file we specify path that spells `...models/insurance_claims/...`. This path agrees with the local path SCv2 was given when it was launched.\n",
    "\n",
    "Now we can deploy all the models with Seldon CLI."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{}\n",
      "{}\n",
      "{}\n",
      "{}\n",
      "{}\n"
     ]
    }
   ],
   "source": [
    "!seldon model load -f ./models/insurance_claims/calculate_claim_value.yaml\n",
    "!seldon model load -f ./models/insurance_claims/classify_claim_value.yaml\n",
    "\n",
    "if use_triton_only:\n",
    "    !seldon model load -f ./models/insurance_claims/classify_claim_complexity.yaml\n",
    "    !seldon model load -f ./models/insurance_claims/calculate_complex_claim_payout.yaml\n",
    "    !seldon model load -f ./models/insurance_claims/calculate_simple_claim_payout.yaml\n",
    "else:\n",
    "    !seldon model load -f ./models/insurance_claims/classify_claim_complexity_mlserver.yaml\n",
    "    !seldon model load -f ./models/insurance_claims/calculate_complex_claim_payout_mlserver.yaml\n",
    "    !seldon model load -f ./models/insurance_claims/calculate_simple_claim_payout_mlserver.yaml"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Similarly, the SCv2 pipeline can be found in the [pipelines](./pipelines) folder. We can output and inspect it here, as it has a number of interesting features."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "apiVersion: mlops.seldon.io/v1alpha1\r\n",
      "kind: Pipeline\r\n",
      "metadata:\r\n",
      "  name: insurance_claims\r\n",
      "  namespace: seldon-mesh\r\n",
      "spec:\r\n",
      "  steps:\r\n",
      "    - name: calculate_claim_value\r\n",
      "      inputs:\r\n",
      "        - insurance_claims.inputs.total_claim_amount\r\n",
      "    - name: classify_claim_value\r\n",
      "      inputs:\r\n",
      "        - calculate_claim_value.outputs.claim_value\r\n",
      "    - name: classify_claim_complexity\r\n",
      "      inputs:\r\n",
      "        - insurance_claims.inputs.total_claim_amount\r\n",
      "        - insurance_claims.inputs.auto_year\r\n",
      "        - insurance_claims.inputs.witnesses\r\n",
      "        - insurance_claims.inputs.police_report_available\r\n",
      "      triggers:\r\n",
      "      - classify_claim_value.outputs.is_low_value_claim\r\n",
      "    - name: calculate_simple_claim_payout\r\n",
      "      inputs:\r\n",
      "      - insurance_claims.inputs.total_claim_amount\r\n",
      "      - insurance_claims.inputs.claim_id\r\n",
      "      triggers:\r\n",
      "      - classify_claim_complexity.outputs.is_simple_claim\r\n",
      "    - name: calculate_complex_claim_payout\r\n",
      "      inputs:\r\n",
      "      - insurance_claims.inputs.total_claim_amount\r\n",
      "      - insurance_claims.inputs.claim_id\r\n",
      "      triggers:\r\n",
      "      - classify_claim_value.outputs.is_high_value_claim\r\n",
      "      - classify_claim_complexity.outputs.is_complex_claim\r\n",
      "      triggersJoinType: any\r\n",
      "  output:\r\n",
      "    steps:\r\n",
      "    - calculate_simple_claim_payout\r\n",
      "    - calculate_complex_claim_payout\r\n",
      "    stepsJoin: any\r\n"
     ]
    }
   ],
   "source": [
    "!cat ./pipelines/insurance_claims.yaml"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "First, note how SCv2 allows us to use pipeline input as an input to any of the pipeline steps. SCv2 allows great flexibility, as any datasteam can be re-use as input of any step.\n",
    "\n",
    "Second, the pipeline uses a lot of triggers. Triggers are special inputs that are not passed to the node themselves, but their presence triggers computation. Triggers can be joined, for example in this pipeline we use `any` join, meanining the computation will be triggered if any of the trigger inputs is present.\n",
    "\n",
    "Third, the pipeline merges outputs together. We have a stream of simple and a stream of complex payouts, and the we merge them into a single pipeline output stream.\n",
    "\n",
    "Now the pipeline can be loaded."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{}\r\n"
     ]
    }
   ],
   "source": [
    "!seldon pipeline load -f ./pipelines/insurance_claims.yaml"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can ensure that models and pipelines are available for us. Here we check their status, which is returned as JSON, and search for specific fields that show their status."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "          \"state\": \"Available\",\r\n",
      "      \"state\": {\r\n",
      "        \"state\": \"ModelAvailable\",\r\n"
     ]
    }
   ],
   "source": [
    "!seldon model status calculate_claim_value | jq -M . | grep state"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "        \"status\": \"PipelineReady\",\r\n"
     ]
    }
   ],
   "source": [
    "!seldon pipeline status insurance_claims | jq -M . | grep status"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Once the pipeline and the models are ready for use, we can issue inference requests to the pipeline. Our input is a CSV file with insurance claims, so we will load it, one row at a time, and form a request JSON string. We can then use CLI to send the request.\n",
    "\n",
    "Below is the helper function to build such request strings. There are actually two utility functions there, as we will be using another one later on in this demo."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "import pandas as pd\n",
    "import base64\n",
    "\n",
    "df = pd.read_csv(\"insurance_claims.csv\")\n",
    "\n",
    "df = df.drop(columns=['incident_date', 'policy_bind_date'])\n",
    "\n",
    "pandas_request_types_map = {\n",
    "    \"int64\": \"INT64\",\n",
    "    \"float64\": \"FP64\",\n",
    "    \"object\": \"BYTES\"\n",
    "}\n",
    "\n",
    "def get_data_value(data, is_rest=False):\n",
    "    if is_rest:\n",
    "        return\n",
    "\n",
    "def get_triton_request_string(claim_index, is_rest=False):\n",
    "    data_key = \"data\" if is_rest else \"contents\"\n",
    "    \n",
    "    records = []\n",
    "    for column_name, column_type in zip(df.columns, df.dtypes):\n",
    "        if not str(column_type) in pandas_request_types_map:\n",
    "            continue\n",
    "\n",
    "        request_type = pandas_request_types_map[str(column_type)]\n",
    "        data = df.iloc[claim_index:claim_index+1, :][column_name].tolist()\n",
    "\n",
    "        content = {\n",
    "            \"name\": column_name,\n",
    "            data_key: data if is_rest else {request_type.lower() + \"_contents\": data},\n",
    "            \"datatype\": request_type,\n",
    "            \"shape\": [1]\n",
    "        }\n",
    "\n",
    "        if column_type == \"object\":\n",
    "            raw_data = content[data_key] if is_rest else content[data_key][\"bytes_contents\"]\n",
    "            base64_data = [base64.b64encode(x.encode()).decode() for x in raw_data]\n",
    "            \n",
    "            if is_rest:\n",
    "                content[data_key] = base64_data\n",
    "            else:\n",
    "                content[data_key][\"bytes_contents\"] = base64_data\n",
    "\n",
    "        records.append(content)\n",
    "\n",
    "    claim_id = {\"name\": \"claim_id\", \"datatype\": \"INT64\", \"shape\": [1]}\n",
    "    claim_id[data_key] = [claim_index] if is_rest else {\"int64_contents\": [claim_index]}\n",
    "    records.append(claim_id)\n",
    "\n",
    "    request = {\"inputs\": records}\n",
    "    if not is_rest:\n",
    "        request[\"model_name\"] = \"does-this-matter?\"\n",
    "\n",
    "    request_string = json.dumps(request)\n",
    "    return request_string\n",
    "\n",
    "def get_mlserver_request_string(claim_index):\n",
    "    records = []\n",
    "    for column_name, column_type in zip(df.columns, df.dtypes):\n",
    "        if not str(column_type) in pandas_request_types_map:\n",
    "            continue\n",
    "\n",
    "        request_type = pandas_request_types_map[str(column_type)]\n",
    "        data = df.iloc[claim_index:claim_index+1, :][column_name].tolist()\n",
    "\n",
    "        content = {\n",
    "            \"name\": column_name,\n",
    "            \"datatype\": request_type,\n",
    "            \"shape\": [1],\n",
    "            \"contents\": {request_type.lower() + \"_contents\": data},\n",
    "        }\n",
    "\n",
    "        if column_type == \"object\":\n",
    "            content[\"contents\"][\"bytes_contents\"] = [base64.b64encode(x.encode()).decode() for x in content[\"contents\"][\"bytes_contents\"]]\n",
    "\n",
    "        records.append(content)\n",
    "\n",
    "    claim_id = {\"name\": \"claim_id\", \"contents\": {\"int64_contents\": [claim_index]}, \"datatype\": \"INT64\", \"shape\": [1]}\n",
    "    records.append(claim_id)\n",
    "    request = {\n",
    "        \"parameters\": {\"content_type\": {\"string_param\": \"pd\"}},\n",
    "        \"model_name\": \"insurance_claims_classifier_1\",\n",
    "        \"inputs\": records\n",
    "    }\n",
    "\n",
    "    request_string = json.dumps(request)\n",
    "    return request_string\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can now send a couple of requests. Response has a handful of information, but for brevity we will only show the numeric output, which is the calculated payout for a given claim."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "          42966\n",
      "          4056\n",
      "          27720\n",
      "          38040\n",
      "          5200\n",
      "          38460\n",
      "          47190\n",
      "          41272\n",
      "          22160\n",
      "          25380\n"
     ]
    }
   ],
   "source": [
    "for i in range(10):\n",
    "    request_string = get_triton_request_string(i)\n",
    "    !seldon pipeline infer insurance_claims --inference-mode grpc '{request_string}' | jq . | grep -A 1 \"fp64Contents\" | grep -v \"fp64Contents\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can also try sending requests via REST"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{\r\n",
      "\t\"model_name\": \"\",\r\n",
      "\t\"outputs\": [\r\n",
      "\t\t{\r\n",
      "\t\t\t\"data\": [\r\n",
      "\t\t\t\t38040\r\n",
      "\t\t\t],\r\n",
      "\t\t\t\"name\": \"claim_payout\",\r\n",
      "\t\t\t\"shape\": [\r\n",
      "\t\t\t\t1,\r\n",
      "\t\t\t\t1\r\n",
      "\t\t\t],\r\n",
      "\t\t\t\"datatype\": \"FP64\"\r\n",
      "\t\t}\r\n",
      "\t]\r\n",
      "}\r\n"
     ]
    }
   ],
   "source": [
    "request_string = get_triton_request_string(3, is_rest=True)\n",
    "!seldon pipeline infer insurance_claims --inference-mode rest '{request_string}'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Stage 2 - data collection"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In the second part of the demo we will emulate a data collection scenario. Suppose we decided to replace the classification part of the pipeline with a simple ML classifier. How can we ago about doing this with SCv2?\n",
    "\n",
    "To train a model, we need to collect a dataset of insurance claims and their complexity. Since SCv2 stores all intermediate computations as data streams, doing this becomes as simple as reading the data off few Kafka streams.\n",
    "\n",
    "We will be reading data from three points in the inference graph:\n",
    "1. Pipeline input stream - that will be the dataset inputs\n",
    "2. Simple claims input stream - labels for simple claims\n",
    "3. Complex claims input stream - labels for complex claims\n",
    "\n",
    "\n",
    "For the purposes of this demo we will use Seldon CLI `inspect` to read the data from the streams, and we start by defining a few utility functions that prcess its outputs. Depending on the use case, you may want to use different tools, such as Kafka clients, to do the same."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 117,
   "metadata": {},
   "outputs": [],
   "source": [
    "def parse_complexity_inspect(lines):\n",
    "    # parses output of\n",
    "    # !seldon pipeline inspect --offset 100 insurance_claims.calculate_complex_claim_payout.inputs\n",
    "\n",
    "    # we only need to collect keys for each record, as mere presence of a record in a stream indicates its complexity\n",
    "    requests = [line.split(\"\\t\", 2)[1] for line in lines]\n",
    "    return requests\n",
    "\n",
    "\n",
    "def parse_pipeline_inputs_inspect(lines):\n",
    "    # parses output of\n",
    "    # !seldon pipeline inspect --offset 100 insurance_claims.inputs\n",
    "    records = []\n",
    "    for line in lines:\n",
    "        tokens = line.split(\"\\t\", 2)\n",
    "        request_id = tokens[1]\n",
    "        request_body = json.loads(tokens[2])\n",
    "\n",
    "        input_record = {}\n",
    "        input_record[\"request_id\"] = request_id\n",
    "        for fields in request_body[\"inputs\"]:\n",
    "            value = next(iter(fields[\"contents\"].values()))[0]\n",
    "            if fields[\"datatype\"] == \"INT64\":\n",
    "                value = int(value)\n",
    "            elif fields[\"datatype\"] == \"FP64\":\n",
    "                value = float(value)\n",
    "            elif fields[\"datatype\"] == \"BYTES\":\n",
    "                value = base64.b64decode(value).decode('utf-8')\n",
    "            input_record[fields[\"name\"]] = value\n",
    "        \n",
    "        records.append(input_record)\n",
    "\n",
    "    return records\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can now read the data and build a dataset."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 118,
   "metadata": {},
   "outputs": [],
   "source": [
    "lines = !seldon pipeline inspect --offset 100 insurance_claims.calculate_complex_claim_payout.inputs\n",
    "complex_requests = parse_complexity_inspect(lines)\n",
    "\n",
    "\n",
    "lines = !seldon pipeline inspect --offset 100 insurance_claims.calculate_simple_claim_payout.inputs\n",
    "simple_requests = parse_complexity_inspect(lines)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 119,
   "metadata": {},
   "outputs": [],
   "source": [
    "lines = !seldon pipeline inspect --offset 100 insurance_claims.inputs\n",
    "pipeline_records = parse_pipeline_inputs_inspect(lines)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 120,
   "metadata": {},
   "outputs": [],
   "source": [
    "for record in pipeline_records:\n",
    "    request_id = record[\"request_id\"]\n",
    "    if request_id in complex_requests:\n",
    "        record[\"is_complex\"] = True\n",
    "    elif request_id in simple_requests:\n",
    "        record[\"is_complex\"] = False\n",
    "    else:\n",
    "        raise ValueError(f\"Request with ID {request_id} seems lost!\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 121,
   "metadata": {},
   "outputs": [],
   "source": [
    "claim_complexity_df = pd.DataFrame(pipeline_records)\n",
    "claim_complexity_df = claim_complexity_df.drop([\"request_id\"], axis='columns')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next step is to build a classification model. We will be using simple Decision Tree classifier from Sciki-learn, and deploy it to MLServer in SCv2. Importantly, to ensure smooth deployment we need to use the same version of Scikit-learn as the one running on the MLServer, which at the moment of writing is 1.1.2."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 122,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "scikit-learn             1.2.0    \n"
     ]
    }
   ],
   "source": [
    "# !pip install scikit-learn==1.2.0\n",
    "\n",
    "!pip list | grep scikit-learn"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Notice that some of our features are strings, while others are numeric. String features have to be encoded, e.g. with one-hot encoding. This can be a separate transformation step in the SCv2 pipeline, but here we will define a pipeline with Scikit-learn instead."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 123,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sklearn.tree import DecisionTreeClassifier\n",
    "from sklearn.preprocessing import OneHotEncoder\n",
    "from sklearn.compose import ColumnTransformer\n",
    "from sklearn.pipeline import Pipeline\n",
    "\n",
    "X = claim_complexity_df.drop(\"is_complex\", axis=1)\n",
    "y = claim_complexity_df[\"is_complex\"]\n",
    "\n",
    "# some columns in the dataset are objects and need to be encoded\n",
    "categorical_features = [column_name for column_name in X.columns if X[column_name].dtype == object]\n",
    "categorical_transformer = OneHotEncoder(handle_unknown=\"ignore\")\n",
    "\n",
    "# other columns are numerical and thus can be just passed through\n",
    "numerical_features = [column_name for column_name in X.columns if X[column_name].dtype != object]\n",
    "\n",
    "\n",
    "preprocessor = ColumnTransformer(\n",
    "    transformers=[\n",
    "        (\"cat\", categorical_transformer, categorical_features),\n",
    "        (\"num\", \"passthrough\", numerical_features),\n",
    "    ]\n",
    ")\n",
    "\n",
    "# max depth here cripples the tree algorithm a bit\n",
    "# so that it isn't perfect\n",
    "classifier = DecisionTreeClassifier(max_depth=2)\n",
    "\n",
    "complete_model = Pipeline(\n",
    "    steps=[(\"preprocessor\", preprocessor), (\"classifier\", classifier)]\n",
    ")\n",
    "\n",
    "complete_model.fit(X, y);"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A simple test run to see what our model outputs."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 124,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "array([1., 0., 0., 1., 0., 1., 1., 0., 0., 1.])"
      ]
     },
     "execution_count": 124,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "complete_model.predict(X)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Stage 3 - pipeline with a model\n",
    "\n",
    "At that stage we have a model that was trained on the data from the pipeline. We are now ready to deploy a new, updated pipeline which makes use of that model."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "First of all, we need to serialize the model. Here we are using joblib for that. Notice that for deployment purposes we expand permissions on the joblib file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 125,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['/home/andrei/projects/insurance-claims-scv2-demo/models/insurance_claims/insurance_claims_classifier/insurance_claims_classifier.joblib']"
      ]
     },
     "execution_count": 125,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import joblib\n",
    "import os\n",
    "\n",
    "base_dir = os.path.abspath('./models/insurance_claims')\n",
    "model_dir = os.path.join(base_dir, \"insurance_claims_classifier\")\n",
    "model_file = os.path.join(model_dir, \"insurance_claims_classifier.joblib\")\n",
    "\n",
    "joblib.dump(complete_model, model_file)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 126,
   "metadata": {},
   "outputs": [],
   "source": [
    "!chmod 666 '{model_file}'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "MLServer deployments require a model settings file that describes the model. It was already prepared, let's have a look."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 127,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{\n",
      "    \"name\": \"insurance_claims_classifier\",\n",
      "    \"implementation\": \"mlserver_sklearn.SKLearnModel\",\n",
      "    \"parameters\":\n",
      "    {\n",
      "        \"uri\": \"./insurance_claims_classifier.joblib\",\n",
      "        \"version\": \"v0.0.1\"\n",
      "    }\n",
      "}"
     ]
    }
   ],
   "source": [
    "model_settings_file = os.path.join(model_dir, \"model-settings.json\")\n",
    "!cat '{model_settings_file}'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As before, we also need a yaml file for the model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 128,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "apiVersion: mlops.seldon.io/v1alpha1\n",
      "kind: Model\n",
      "metadata:\n",
      "  name: insurance_claims_classifier\n",
      "  namespace: seldon-mesh\n",
      "spec:\n",
      "  storageUri: \"/mnt/models/insurance_claims/insurance_claims_classifier\"\n",
      "  requirements:\n",
      "  - sklearn\n"
     ]
    }
   ],
   "source": [
    "model_seldon_file = os.path.join(base_dir, \"insurance_claims_classifier.yaml\")\n",
    "!cat '{model_seldon_file}'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "At the moment SCv2 triggers only work on presence or absence of a record. Whereas our model outputs boolean values. So we also need a simple fork, that takes model ouputs and routes it to one stream or another, based on value. We call this model `is_complex_conditional`, and it can be inspected in the models folder. Here we upload it to SCv2 along with the classifier."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 129,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{}\n",
      "{}\n"
     ]
    }
   ],
   "source": [
    "!seldon model load -f '{model_seldon_file}'\n",
    "\n",
    "if use_triton_only:\n",
    "    !seldon model load -f ./models/insurance_claims/is_complex_conditional.yaml\n",
    "else:\n",
    "    !seldon model load -f ./models/insurance_claims/is_complex_conditional_mlserver.yaml"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 130,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "          \"state\": \"Available\",\n",
      "      \"state\": {\n",
      "        \"state\": \"ModelAvailable\",\n"
     ]
    }
   ],
   "source": [
    "!seldon model status insurance_claims_classifier | jq -M . | grep state"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we need a new pipeline. You can see that we are reusing some of the steps from the previous pipeline."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 131,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "apiVersion: mlops.seldon.io/v1alpha1\n",
      "kind: Pipeline\n",
      "metadata:\n",
      "  name: insurance_claims_with_model\n",
      "  namespace: seldon-mesh\n",
      "spec:\n",
      "  steps:\n",
      "    - name: insurance_claims_classifier\n",
      "      inputs:\n",
      "      - insurance_claims_with_model.inputs\n",
      "    - name: is_complex_conditional\n",
      "      inputs:\n",
      "      - insurance_claims_classifier.outputs.predict\n",
      "      tensorMap:\n",
      "        insurance_claims_classifier.outputs.predict: is_complex\n",
      "    - name: calculate_simple_claim_payout\n",
      "      inputs:\n",
      "      - insurance_claims_with_model.inputs.total_claim_amount\n",
      "      - insurance_claims_with_model.inputs.claim_id\n",
      "      triggers:\n",
      "      - is_complex_conditional.outputs.is_simple_claim\n",
      "    - name: calculate_complex_claim_payout\n",
      "      inputs:\n",
      "      - insurance_claims_with_model.inputs.total_claim_amount\n",
      "      - insurance_claims_with_model.inputs.claim_id\n",
      "      triggers:\n",
      "      - is_complex_conditional.outputs.is_complex_claim\n",
      "  output:\n",
      "    steps:\n",
      "    - calculate_simple_claim_payout\n",
      "    - calculate_complex_claim_payout\n",
      "    stepsJoin: any\n",
      "\n",
      "\n",
      "\n"
     ]
    }
   ],
   "source": [
    "!cat ./pipelines/insurance_claims_with_model.yaml"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 132,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{}\n"
     ]
    }
   ],
   "source": [
    "!seldon pipeline load -f ./pipelines/insurance_claims_with_model.yaml"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 133,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "        \"status\": \"PipelineReady\",\n"
     ]
    }
   ],
   "source": [
    "!seldon pipeline status insurance_claims_with_model | jq -M . | grep status"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "With the new model and new pipeline ready, we can do the requests again."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 134,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "          42966\n",
      "          4056\n",
      "          27720\n",
      "          38040\n",
      "          5200\n",
      "          38460\n",
      "          47190\n",
      "          41272\n",
      "          22160\n",
      "          25380\n"
     ]
    }
   ],
   "source": [
    "for i in range(10):\n",
    "    request_string = get_mlserver_request_string(i)\n",
    "    !seldon pipeline infer insurance_claims_with_model --inference-mode grpc '{request_string}' | jq . | grep -A 1 \"fp64Contents\" | grep -v \"fp64Contents\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "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.8.10"
  },
  "vscode": {
   "interpreter": {
    "hash": "19cc22dabd606a593ba102d44a99a75eef7b9a8fc3a94de330ab35411bc896ad"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}