notebooks/ch_08_model_serving.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "36d84a44",
   "metadata": {},
   "source": [
    "# Online Inference with Ray Serve"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dcc579d9",
   "metadata": {},
   "source": [
    "\n",
    "You can run this notebook directly in\n",
    "[Colab](https://colab.research.google.com/github/maxpumperla/learning_ray/blob/main/notebooks/ch_08_model_serving.ipynb).\n",
    "<a target=\"_blank\" href=\"https://colab.research.google.com/github/maxpumperla/learning_ray/blob/main/notebooks/ch_08_model_serving.ipynb\">\n",
    "<img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1545f53d",
   "metadata": {},
   "source": [
    "For this chapter you need to install the following dependencies:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9633bd83",
   "metadata": {},
   "outputs": [],
   "source": [
    "! pip install \"ray[serve]==2.2.0\" \"transformers==4.21.2\"\n",
    "! pip install \"requests==2.28.1\" \"wikipedia==1.4.0\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b9d2fed7",
   "metadata": {},
   "source": [
    "\n",
    "To import utility files for this chapter, on Colab you will also have to clone\n",
    "the repo and copy the code files to the base path of the runtime:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "34fe8c35",
   "metadata": {},
   "outputs": [],
   "source": [
    "!git clone https://github.com/maxpumperla/learning_ray\n",
    "%cp -r learning_ray/notebooks/* ."
   ]
  },
  {
   "cell_type": "markdown",
   "source": [
    "![Serve Positioning](https://raw.githubusercontent.com/maxpumperla/learning_ray/main/notebooks/images/chapter_08/serve_positioning.png)"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "markdown",
   "source": [
    "![Serve Architecture](https://raw.githubusercontent.com/maxpumperla/learning_ray/main/notebooks/images/chapter_08/serve_arch.png)"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "markdown",
   "source": [
    "![NLP API Architecture](https://raw.githubusercontent.com/maxpumperla/learning_ray/main/notebooks/images/chapter_08/nlp_api_arch.png)"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c18b1864",
   "metadata": {},
   "outputs": [],
   "source": [
    "from ray import serve\n",
    "\n",
    "from transformers import pipeline\n",
    "\n",
    "\n",
    "@serve.deployment\n",
    "class SentimentAnalysis:\n",
    "    def __init__(self):\n",
    "        self._classifier = pipeline(\"sentiment-analysis\")\n",
    "\n",
    "    def __call__(self, request) -> str:\n",
    "        input_text = request.query_params[\"input_text\"]\n",
    "        return self._classifier(input_text)[0][\"label\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "60908300",
   "metadata": {},
   "outputs": [],
   "source": [
    "basic_deployment = SentimentAnalysis.bind()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b7659c67",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Run this in a separate process to avoid any blocking:\n",
    "! serve run --non-blocking app:basic_deployment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "042a810c",
   "metadata": {
    "lines_to_next_cell": 2
   },
   "outputs": [],
   "source": [
    "import requests\n",
    "\n",
    "print(requests.get(\n",
    "    \"http://localhost:8000/\", params={\"input_text\": \"Hello friend!\"}\n",
    ").json())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8452a0b4",
   "metadata": {
    "lines_to_next_cell": 2
   },
   "outputs": [],
   "source": [
    "from fastapi import FastAPI\n",
    "\n",
    "app = FastAPI()\n",
    "\n",
    "\n",
    "@serve.deployment\n",
    "@serve.ingress(app)\n",
    "class SentimentAnalysis:\n",
    "    def __init__(self):\n",
    "        self._classifier = pipeline(\"sentiment-analysis\")\n",
    "\n",
    "    @app.get(\"/\")\n",
    "    def classify(self, input_text: str) -> str:\n",
    "        return self._classifier(input_text)[0][\"label\"]\n",
    "\n",
    "\n",
    "fastapi_deployment = SentimentAnalysis.bind()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "987200e3",
   "metadata": {
    "lines_to_next_cell": 2
   },
   "outputs": [],
   "source": [
    "app = FastAPI()\n",
    "\n",
    "\n",
    "@serve.deployment(num_replicas=2, ray_actor_options={\"num_cpus\": 2})\n",
    "@serve.ingress(app)\n",
    "class SentimentAnalysis:\n",
    "    def __init__(self):\n",
    "        self._classifier = pipeline(\"sentiment-analysis\")\n",
    "\n",
    "    @app.get(\"/\")\n",
    "    def classify(self, input_text: str) -> str:\n",
    "        import os\n",
    "        print(\"from process:\", os.getpid())\n",
    "        return self._classifier(input_text)[0][\"label\"]\n",
    "\n",
    "\n",
    "scaled_deployment = SentimentAnalysis.bind()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "28b7996d",
   "metadata": {},
   "outputs": [],
   "source": [
    "app = FastAPI()\n",
    "\n",
    "\n",
    "@serve.deployment\n",
    "@serve.ingress(app)\n",
    "class SentimentAnalysis:\n",
    "    def __init__(self):\n",
    "        self._classifier = pipeline(\"sentiment-analysis\")\n",
    "\n",
    "    @serve.batch(max_batch_size=10, batch_wait_timeout_s=0.1)\n",
    "    async def classify_batched(self, batched_inputs):\n",
    "        print(\"Got batch size:\", len(batched_inputs))\n",
    "        results = self._classifier(batched_inputs)\n",
    "        return [result[\"label\"] for result in results]\n",
    "\n",
    "    @app.get(\"/\")\n",
    "    async def classify(self, input_text: str) -> str:\n",
    "        return await self.classify_batched(input_text)\n",
    "\n",
    "\n",
    "batched_deployment = SentimentAnalysis.bind()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a167c10b",
   "metadata": {
    "lines_to_next_cell": 2
   },
   "outputs": [],
   "source": [
    "import ray\n",
    "from ray import serve\n",
    "from app import batched_deployment\n",
    "\n",
    "handle = serve.run(batched_deployment)\n",
    "ray.get([handle.classify.remote(\"sample text\") for _ in range(10)])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c6eb34fe",
   "metadata": {
    "lines_to_next_cell": 2
   },
   "outputs": [],
   "source": [
    "@serve.deployment\n",
    "class DownstreamModel:\n",
    "    def __call__(self, inp: str):\n",
    "        return \"Hi from downstream model!\"\n",
    "\n",
    "\n",
    "@serve.deployment\n",
    "class Driver:\n",
    "    def __init__(self, downstream):\n",
    "        self._d = downstream\n",
    "\n",
    "    async def __call__(self, *args) -> str:\n",
    "        return await self._d.remote()\n",
    "\n",
    "\n",
    "downstream = DownstreamModel.bind()\n",
    "driver = Driver.bind(downstream)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e56ac5a3",
   "metadata": {
    "lines_to_next_cell": 2
   },
   "outputs": [],
   "source": [
    "@serve.deployment\n",
    "class DownstreamModel:\n",
    "    def __init__(self, my_val: str):\n",
    "        self._my_val = my_val\n",
    "\n",
    "    def __call__(self, inp: str):\n",
    "        return inp + \"|\" + self._my_val\n",
    "\n",
    "\n",
    "@serve.deployment\n",
    "class PipelineDriver:\n",
    "    def __init__(self, model1, model2):\n",
    "        self._m1 = model1\n",
    "        self._m2 = model2\n",
    "\n",
    "    async def __call__(self, *args) -> str:\n",
    "        intermediate = self._m1.remote(\"input\")\n",
    "        final = self._m2.remote(intermediate)\n",
    "        return await final\n",
    "\n",
    "\n",
    "m1 = DownstreamModel.bind(\"val1\")\n",
    "m2 = DownstreamModel.bind(\"val2\")\n",
    "pipeline_driver = PipelineDriver.bind(m1, m2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7a5cfea3",
   "metadata": {
    "lines_to_next_cell": 2
   },
   "outputs": [],
   "source": [
    "@serve.deployment\n",
    "class DownstreamModel:\n",
    "    def __init__(self, my_val: str):\n",
    "        self._my_val = my_val\n",
    "\n",
    "    def __call__(self):\n",
    "        return self._my_val\n",
    "\n",
    "\n",
    "@serve.deployment\n",
    "class BroadcastDriver:\n",
    "    def __init__(self, model1, model2):\n",
    "        self._m1 = model1\n",
    "        self._m2 = model2\n",
    "\n",
    "    async def __call__(self, *args) -> str:\n",
    "        output1, output2 = self._m1.remote(), self._m2.remote()\n",
    "        return [await output1, await output2]\n",
    "\n",
    "\n",
    "m1 = DownstreamModel.bind(\"val1\")\n",
    "m2 = DownstreamModel.bind(\"val2\")\n",
    "broadcast_driver = BroadcastDriver.bind(m1, m2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "26b08ba4",
   "metadata": {
    "lines_to_next_cell": 2
   },
   "outputs": [],
   "source": [
    "@serve.deployment\n",
    "class DownstreamModel:\n",
    "    def __init__(self, my_val: str):\n",
    "        self._my_val = my_val\n",
    "\n",
    "    def __call__(self):\n",
    "        return self._my_val\n",
    "\n",
    "\n",
    "@serve.deployment\n",
    "class ConditionalDriver:\n",
    "    def __init__(self, model1, model2):\n",
    "        self._m1 = model1\n",
    "        self._m2 = model2\n",
    "\n",
    "    async def __call__(self, *args) -> str:\n",
    "        import random\n",
    "        if random.random() > 0.5:\n",
    "            return await self._m1.remote()\n",
    "        else:\n",
    "            return await self._m2.remote()\n",
    "\n",
    "\n",
    "m1 = DownstreamModel.bind(\"val1\")\n",
    "m2 = DownstreamModel.bind(\"val2\")\n",
    "conditional_driver = ConditionalDriver.bind(m1, m2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "33ed39cb",
   "metadata": {},
   "outputs": [],
   "source": [
    "from typing import Optional\n",
    "\n",
    "import wikipedia\n",
    "\n",
    "\n",
    "def fetch_wikipedia_page(search_term: str) -> Optional[str]:\n",
    "    results = wikipedia.search(search_term)\n",
    "    # If no results, return to caller.\n",
    "    if len(results) == 0:\n",
    "        return None\n",
    "\n",
    "    # Get the page for the top result.\n",
    "    return wikipedia.page(results[0]).content"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2646606f",
   "metadata": {},
   "outputs": [],
   "source": [
    "from ray import serve\n",
    "from transformers import pipeline\n",
    "from typing import List\n",
    "\n",
    "\n",
    "@serve.deployment\n",
    "class SentimentAnalysis:\n",
    "    def __init__(self):\n",
    "        self._classifier = pipeline(\"sentiment-analysis\")\n",
    "\n",
    "    @serve.batch(max_batch_size=10, batch_wait_timeout_s=0.1)\n",
    "    async def is_positive_batched(self, inputs: List[str]) -> List[bool]:\n",
    "        results = self._classifier(inputs, truncation=True)\n",
    "        return [result[\"label\"] == \"POSITIVE\" for result in results]\n",
    "\n",
    "    async def __call__(self, input_text: str) -> bool:\n",
    "        return await self.is_positive_batched(input_text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5d37f206",
   "metadata": {},
   "outputs": [],
   "source": [
    "@serve.deployment(num_replicas=2)\n",
    "class Summarizer:\n",
    "    def __init__(self, max_length: Optional[int] = None):\n",
    "        self._summarizer = pipeline(\"summarization\")\n",
    "        self._max_length = max_length\n",
    "\n",
    "    def __call__(self, input_text: str) -> str:\n",
    "        result = self._summarizer(\n",
    "            input_text, max_length=self._max_length, truncation=True)\n",
    "        return result[0][\"summary_text\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "931685ff",
   "metadata": {},
   "outputs": [],
   "source": [
    "@serve.deployment\n",
    "class EntityRecognition:\n",
    "    def __init__(self, threshold: float = 0.90, max_entities: int = 10):\n",
    "        self._entity_recognition = pipeline(\"ner\")\n",
    "        self._threshold = threshold\n",
    "        self._max_entities = max_entities\n",
    "\n",
    "    def __call__(self, input_text: str) -> List[str]:\n",
    "        final_results = []\n",
    "        for result in self._entity_recognition(input_text):\n",
    "            if result[\"score\"] > self._threshold:\n",
    "                final_results.append(result[\"word\"])\n",
    "            if len(final_results) == self._max_entities:\n",
    "                break\n",
    "\n",
    "        return final_results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4b925bb1",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pydantic import BaseModel\n",
    "\n",
    "\n",
    "class Response(BaseModel):\n",
    "    success: bool\n",
    "    message: str = \"\"\n",
    "    summary: str = \"\"\n",
    "    named_entities: List[str] = []"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "479767a8",
   "metadata": {},
   "outputs": [],
   "source": [
    "from fastapi import FastAPI\n",
    "\n",
    "app = FastAPI()\n",
    "\n",
    "\n",
    "@serve.deployment\n",
    "@serve.ingress(app)\n",
    "class NLPPipelineDriver:\n",
    "    def __init__(self, sentiment_analysis, summarizer, entity_recognition):\n",
    "        self._sentiment_analysis = sentiment_analysis\n",
    "        self._summarizer = summarizer\n",
    "        self._entity_recognition = entity_recognition\n",
    "\n",
    "    @app.get(\"/\", response_model=Response)\n",
    "    async def summarize_article(self, search_term: str) -> Response:\n",
    "        # Fetch the top page content for the search term if found.\n",
    "        page_content = fetch_wikipedia_page(search_term)\n",
    "        if page_content is None:\n",
    "            return Response(success=False, message=\"No pages found.\")\n",
    "\n",
    "        # Conditionally continue based on the sentiment analysis.\n",
    "        is_positive = await self._sentiment_analysis.remote(page_content)\n",
    "        if not is_positive:\n",
    "            return Response(success=False, message=\"Only positivitiy allowed!\")\n",
    "\n",
    "        # Query the summarizer and named entity recognition models in parallel.\n",
    "        summary_result = self._summarizer.remote(page_content)\n",
    "        entities_result = self._entity_recognition.remote(page_content)\n",
    "        return Response(\n",
    "            success=True,\n",
    "            summary=await summary_result,\n",
    "            named_entities=await entities_result\n",
    "        )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b7ec00f5",
   "metadata": {},
   "outputs": [],
   "source": [
    "sentiment_analysis = SentimentAnalysis.bind()\n",
    "summarizer = Summarizer.bind()\n",
    "entity_recognition = EntityRecognition.bind(threshold=0.95, max_entities=5)\n",
    "nlp_pipeline_driver = NLPPipelineDriver.bind(\n",
    "    sentiment_analysis, summarizer, entity_recognition)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "62408257",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Run this in a separate process to avoid any blocking:\n",
    "! serve run --non-blocking app:nlp_pipeline_driver"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f456cfc9",
   "metadata": {
    "lines_to_next_cell": 2
   },
   "outputs": [],
   "source": [
    "import requests\n",
    "\n",
    "\n",
    "print(requests.get(\n",
    "    \"http://localhost:8000/\", params={\"search_term\": \"rayserve\"}\n",
    ").text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6448177a",
   "metadata": {
    "lines_to_next_cell": 2
   },
   "outputs": [],
   "source": [
    "print(requests.get(\n",
    "    \"http://localhost:8000/\", params={\"search_term\": \"war\"}\n",
    ").text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e0b6dccf",
   "metadata": {
    "lines_to_next_cell": 2
   },
   "outputs": [],
   "source": [
    "print(requests.get(\n",
    "    \"http://localhost:8000/\", params={\"search_term\": \"physicist\"}\n",
    ").text)"
   ]
  }
 ],
 "metadata": {
  "jupytext": {
   "cell_metadata_filter": "-all",
   "main_language": "python",
   "notebook_metadata_filter": "-all"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}