examples/openai-spam-detection/openai_spam_detection.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "7738ae81-1fcc-46ea-a8ca-d80497becbe6",
   "metadata": {},
   "source": [
    "# Spam Detection with OpenAI\n",
    "\n",
    "This example showcases how to leverage OpenAI within the transform function and update the events data with the response. \n",
    "In this example, we will detect spam comments using OpenAI and add the result to the event. \n",
    "\n",
    "The entire process runs as a seververless event driven pipeline on GlassFlow. \n",
    "\n",
    "## Pre-requisites\n",
    "\n",
    "- Create your free GlassFlow account via the [GlassFlow WebApp](https://app.glassflow.dev).\n",
    "- Get your [Personal Access Token](https://app.glassflow.dev/profile) to authorize the Python SDK to interact with GlassFlow Cloud.\n",
    "- Get your OpenAI API Key https://platform.openai.com/"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c9aea43d-888a-49bf-a878-34a059f1849b",
   "metadata": {},
   "outputs": [],
   "source": "%pip install \"glassflow>=2.0.8\" pandas"
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "21fe134e-87db-49ce-b402-233c70b2d6a3",
   "metadata": {},
   "outputs": [],
   "source": [
    "import glassflow"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5ea2eff3-4f22-4046-8f71-77844b4e283b",
   "metadata": {},
   "outputs": [],
   "source": [
    "# fill credentials\n",
    "# Please edit this variable with your own personal access token from https://app.glassflow.dev/profile\n",
    "personal_access_token = \"\"\n",
    "OPENAI_API_KEY=\"\"\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "85f568b5-dd14-4fbf-88cd-b6d43eb3b855",
   "metadata": {},
   "source": [
    "## Create Pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "774b8473-64c0-41f9-89aa-9c553bbc4869",
   "metadata": {},
   "outputs": [],
   "source": [
    "client = glassflow.GlassFlowClient(\n",
    "    personal_access_token=personal_access_token\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "21760fbe-399e-43d3-8e71-887a01674eec",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get the space named \"examples\" (or create one if no space is found)\n",
    "list_spaces = client.list_spaces()\n",
    "\n",
    "space_name = \"examples\"\n",
    "for s in list_spaces.spaces:\n",
    "    if s[\"name\"] == space_name:\n",
    "        space = glassflow.Space(\n",
    "            personal_access_token=client.personal_access_token,\n",
    "            id=s[\"id\"], \n",
    "            name=s[\"name\"]\n",
    "        )\n",
    "        break\n",
    "else:\n",
    "    space = client.create_space(name=space_name)\n",
    "\n",
    "print(f\"Space \\\"{space.name}\\\" with ID: {space.id}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "10e62973-a501-48c8-9c5b-64eb17f4376a",
   "metadata": {},
   "source": [
    "### Transformation Function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "62e7cc28-57af-499b-a1ee-e92cb6a366ba",
   "metadata": {},
   "outputs": [],
   "source": [
    "%pycat transform.py"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c76c9f5f-5f81-4536-97c3-8d8a566baf80",
   "metadata": {},
   "source": [
    "### Requirements txt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c080edf1-2328-4b45-96fe-476e5b4d309b",
   "metadata": {},
   "outputs": [],
   "source": [
    "with open(\"requirements.txt\") as f:\n",
    "    requirements_txt = f.read()\n",
    "print(requirements_txt)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "73c44e6e-322a-4caa-bf3f-65f4ef6f11f7",
   "metadata": {},
   "source": [
    "### Environment variables"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6f8aee2b-787d-43d7-8db5-b32583db9ce4",
   "metadata": {},
   "outputs": [],
   "source": [
    "env_vars = [{\n",
    "  \"name\": \"OPENAI_API_KEY\",\n",
    "  \"value\": OPENAI_API_KEY\n",
    "}]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a7a01e89-2d31-4dcf-b854-782003051cd8",
   "metadata": {},
   "source": [
    "### Create Pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d473940b-0935-40bc-b8b8-d6afad98e4f6",
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline_name = \"spam-detection-openai\"\n",
    "\n",
    "pipeline = client.create_pipeline(\n",
    "    name=pipeline_name, \n",
    "    transformation_file='transform.py',\n",
    "    space_id=space.id, \n",
    "    env_vars=env_vars,\n",
    "    requirements=requirements_txt\n",
    ")\n",
    "print(\"Pipeline ID:\", pipeline.id)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c46a5402-86b8-4184-a01c-f052b6684374",
   "metadata": {},
   "outputs": [],
   "source": [
    "print(\"Pipeline is deployed!\") \n",
    "print(\"Pipeline Id = %s\" % (pipeline.id))\n",
    "print(\"Pipeline URL %s \"% f\"https://app.glassflow.dev/pipelines/{pipeline.id}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "406c9a08-9551-4e28-9e1b-da2293286fa4",
   "metadata": {},
   "source": [
    "## Produce data and send it to your pipeline"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7fb99f74-aeb5-4ac0-8b51-c4aad0ce06b3",
   "metadata": {},
   "source": [
    "### Get sample data from a public dataset "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "26d3ace9-ca13-44d1-a834-24c43fb3015e",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "\n",
    "def get_data_sample(sample_size=100):\n",
    "    \"\"\"\n",
    "    Fetches and samples the SMS spam dataset.\n",
    "    \"\"\"\n",
    "    url = 'https://raw.githubusercontent.com/justmarkham/DAT8/master/data/sms.tsv'\n",
    "    df = pd.read_csv(url, sep='\\t', header=None, names=['label', 'message'])\n",
    "    # Sample specified number of negative and positive examples without replacement\n",
    "    negatives = df[df['label'] == 'ham'].sample(sample_size//2, random_state=42)\n",
    "    positives = df[df['label'] == 'spam'].sample(sample_size//2, random_state=42)\n",
    "    df_sampled = pd.concat([negatives, positives]).reset_index(drop=True)\n",
    "    df_shuffled = df_sampled.sample(frac=1)\n",
    "    data = df_shuffled.to_dict('records')\n",
    "    return data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "88301804-47c5-4d8f-94ca-de3236ace858",
   "metadata": {},
   "outputs": [],
   "source": [
    "sample_chat_dataset = get_data_sample()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c65cfa9e-5bbc-4a12-8fd1-87d9e568b472",
   "metadata": {},
   "source": [
    "### Get pipeline data source object to publish events to the pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a0916568-2172-4f5c-a309-6a0b571073d5",
   "metadata": {},
   "outputs": [],
   "source": [
    "data_source = pipeline.get_source()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2f2ec250-a783-4bef-bfe0-61e538649a61",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Generate some data and send it to the pipeline. Store it locally to compare\n",
    "n_events = 10\n",
    "input_events = []\n",
    "for event in sample_chat_dataset[0:n_events]:    \n",
    "    input_events.append(event)\n",
    "    data_source.publish(event)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "70a99d5f-2783-4784-8792-389ac55a114f",
   "metadata": {},
   "source": [
    "### Display data sent to the pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fb01915a-be2b-467b-9f76-51150eaa7f97",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "\n",
    "display(pd.DataFrame(input_events))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "91e5b991-4727-4c0f-8cc1-2e39d54566a0",
   "metadata": {},
   "source": [
    "## Consume events from the pipeline \n",
    "\n",
    "Get pipeline data sink to consume the transformed events from the pipeline."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "27aee4a0-57d1-4732-9e49-72638e298f12",
   "metadata": {},
   "outputs": [],
   "source": [
    "data_sink = pipeline.get_sink()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a6466065-4a59-4ae4-be16-b8c8ed381127",
   "metadata": {},
   "outputs": [],
   "source": [
    "output_events = []\n",
    "while True:\n",
    "    resp = data_sink.consume()\n",
    "    if resp.status_code == 200:\n",
    "        output_events.append(resp.json())\n",
    "    if len(output_events) == n_events:\n",
    "        # all events have been consumed\n",
    "        break"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "81fc920b-d5f2-405d-8e88-e2b51b37f797",
   "metadata": {},
   "source": [
    "### Display data received from the pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "61a2f6cc-5b4d-46bd-bd64-76f80658af23",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "\n",
    "display(pd.DataFrame(output_events))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "184f7e4d-9121-43ad-b587-5e8eea794510",
   "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.9.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}