Tutorials/Tutorials_DIY/1.ReverseNumber/ReverseNumber.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Tutorial 1\n",
    "\n",
    "**In this tutorial you will:**\n",
    "- learn the Flow basics and implement your first Flow [Section 1](#1-writing-your-first-atomicFlow-reversenumberatomicflow)\n",
    "- understand the basics of Flows engine and serve your first Flow [Section 2](#2-Flow-engine)\n",
    "- learn how to call served Flows [Section 2](#24-getting-an-instance-of-the-served-flow-via-a-proxy)\n",
    "- learn how to compose Flows into more complex Flows [Section 3](#3writing-your-first-compositeFlow-reversenumbersequentialflow)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%load_ext autoreload\n",
    "%autoreload 2\n",
    "#imports\n",
    "from aiflows.utils.general_helpers import read_yaml_file\n",
    "from aiflows.utils import serving\n",
    "from aiflows.utils import colink_utils\n",
    "from aiflows.workers import run_dispatch_worker_thread\n",
    "from aiflows.base_flows import AtomicFlow\n",
    "from aiflows.messages import FlowMessage\n",
    "import sys\n",
    "import os\n",
    "sys.path.append(\"..\")\n",
    "from utils import compile_and_writefile, dict_to_yaml\n",
    "\n",
    "# Specify path of your Flow modules\n",
    "FLOW_MODULES_PATH = \"./\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. Writing your First AtomicFlow: ReverseNumberAtomicFlow\n",
    "\n",
    "In this section we will introduce the concept of a Flow and implement a simple Flow that reverses a number passed in the input message.\n",
    "\n",
    "In a nutshell, Flows are objects that perform semantically meaningful unit of work and comunicate via messages.\n",
    "\n",
    "There are two types of Flows: \n",
    "1) **Atomic Flows:** Performs the computation directly\n",
    "2) **Composite Flows:** Orchestrates other Flows in performing the computation\n",
    "\n",
    "When a message is send to a Flow, the Flow processes the message according to the logic specified in its `run` method. We will now go through the implementation of the `ReverseNumberAtomicFlow`: a Flow who given a number, reverses it (e.g., 12345-> 54321).\n",
    "\n",
    "In Section 3, we see provide an example implementation of a Composite Flow."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1.1 Writing the ReverseNumberAtomicFlow\n",
    "\n",
    "To implement a Flow, you must define its run method. This method takes an input_message as its input, which is an instance of the FlowMessage class. Within the FlowMessage, there are two important attributes to consider:\n",
    "\n",
    " - 1. `data`: This attribute is a dictionary containing the data that the Flow will process.\n",
    " - 2. `reply_data`: Another dictionary, it holds necessary information to send a reply message back to the sender of the input message. Typically, direct modification of this attribute isn't recommended. Instead, utilize the provided methods for modifying it (these methods will be introduced throughout the notebook).\n",
    "\n",
    "When defining a Flow, it's crucial to specify the `input_interface` and `output_interface` in the configuration file. These interfaces specify the expected keys in the `data` attribute of the `input_message` and what will be returned in the `data` attribute when sending the message back. This ensures proper calling of your Flow by other users. For instance, in the `ReverseNumberAtomicFlow`, the `input_interface` expects a key `number` in the `data` attribute of the `input_message`, while the `output_interface` will include a key `reversed_number` in the `data` attribute of the sent message. Once the number is reversed, package the response dictionary into a message using the `package_output_message` method. This method packages the dictionary into a message and adjusts the `reply_data` attribute of the message to match that of the input message, ensuring it is sent back to the original sender. Finally, the reply is sent back using the `send_message` method.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%compile_and_writefile ./ReverseNumberFlowModule/ReverseNumberAtomicFlow.py\n",
    "\n",
    "\n",
    "from aiflows.base_flows import AtomicFlow\n",
    "from aiflows.messages import FlowMessage\n",
    "\n",
    "class ReverseNumberAtomicFlow(AtomicFlow):\n",
    "    \n",
    "    # Customize the logic within this function as needed for your specific Flow requirements.\n",
    "    def run(self, input_message: FlowMessage):\n",
    "        ## ~~~~ Getting the input ~~~~\n",
    "        # Get the input dictionary from the input message\n",
    "        input_data = input_message.data\n",
    "        \n",
    "        # Get input number from data dictionary\n",
    "        input_number = input_data[\"number\"]\n",
    "        \n",
    "        ## ~~~~ Main logic ~~~~\n",
    "        ########## YOUR CODE GOES HERE ##########\n",
    "        # TODO: Reverse the input number (e.g. 1234 -> 4321)\n",
    "        reversed_number = int(str(input_number)[::-1])\n",
    "        ########## YOUR CODE GOES HERE ##########\n",
    "        \n",
    "        # ~~~ Preparing the response message ~~~\n",
    "        response = {\"reversed_number\": reversed_number}\n",
    "        \n",
    "        # This method packages the `response` in a FlowMessage object \n",
    "        # containing the necessary metadata to send the message back\n",
    "        # to the sender of the input message  \n",
    "        reply = self.package_output_message(\n",
    "            input_message=input_message,\n",
    "            response=response,\n",
    "        )\n",
    "        \n",
    "        # ~~~ Sending the response ~~~\n",
    "        self.send_message(\n",
    "            reply\n",
    "        )\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note: Please be aware that the cell magic `compile_and_writefile` is saving the cell above at `./ReverseNumberFlowModule/ReverseNumberAtomicFlow.py`.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "### 1.2 Creating the Default Configuration File for the Reverse Number Flow Class\n",
    "\n",
    "For each Flow class, there exists a corresponding default configuration file. This file delineates the parameters to be configured for the Flow. Access to each field in the configuration file is facilitated within the Flow class through the syntax `self.flow_config[\"FIELD_NAME\"]`. For instance, to retrieve the `input_interface` field, you would utilize `self.flow_config[\"input_interface\"]`.\n",
    "\n",
    "Now, let's proceed with composing the configuration file for the `ReverseNumberAtomicFlow`. It's noteworthy that we define it as a dictionary here and subsequently save it to a `.yaml` file. However, you also have the option to directly craft the configuration within a `.yaml` file. \n",
    "\n",
    "**IMPORTANT: Ensure that the configuration file is named identical to the Flow class name and is placed in the same directory as the Flow. If you don't, it won't be considered as the default configuration file of that Flow.**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "default_config_reverse_number = \\\n",
    "{\n",
    "    \"name\": \"ReverseNumber\",\n",
    "    \"description\": \"A Flow that takes in a number and reverses it.\",\n",
    "\n",
    "    \"_target_\": \"ReverseNumberFlowModule.ReverseNumberAtomicFlow.ReverseNumberAtomicFlow.instantiate_from_default_config\",\n",
    "\n",
    "    \"input_interface\": \"number\",\n",
    "    \"output_interface\": \"reversed_number\",\n",
    "}\n",
    "dict_to_yaml(default_config_reverse_number, \"./ReverseNumberFlowModule/ReverseNumberAtomicFlow.yaml\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's dissect the configuration file for the `ReverseNumberAtomicFlow`:\n",
    "\n",
    "- **`name` and `description`**: These parameters are straightforward. When defining a Flow, these must always be specified. Although they don't directly affect the functionality of the Flow, they are essential for documentation purposes.\n",
    "\n",
    "- **`_target_`**: This parameter specifies how the Flow will be instantiated by `hydra`, a framework for configuring complex applications. It should always specify the path to the Flow class followed by `instantiate_from_default_config`. This standardizes the instantiation process from the default configuration file. For example, the value of `_target_` is `ReverseNumberFlowModule.ReverseNumberAtomic.ReverseNumberAtomicFlow.instantiate_from_default_config`, where `ReverseNumberFlowModule/ReverseNumberAtomic.py` is the path to the Flow class, and the class name in `ReverseNumberAtomic.py` is `ReverseNumberAtomic`.\n",
    "\n",
    "- **`input_interface`**: This parameter outlines the keys expected in the `data` attribute of the input message. Here, the `number` key is expected.\n",
    "\n",
    "- **`output_interface`**: Specifies the keys present in the `data` attribute of the message sent back. In this case, the `reversed_number` key will be included."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. Flow Engine\n",
    "\n",
    "In this section we introduce the basics of the Flows Engine, and guide you through the process of serving and accessing served Flows.\n",
    "\n",
    "The Flows Engine is build on top of [CoLink](https://github.com/CoLearn-Dev/colink-sdk-python-dev). As mentioned in their documentation:\n",
    "\n",
    "> CoLink provides a unified interface for the user, storage, communication, and computation. Extending gRPC, CoLink simplifies the development of multi-party protocols and allow implementations in different programming languages to work together consistently. With a unified interface that increases potential data contributors, CoLink has the potential to enable larger-scale decentralized data collaboration and unlock the true value of data.\n",
    "\n",
    "In simpler terms, CoLink allows aiflows to run multiple Flows, which can be spread across multiple machines to complete a task. It enables communication between Flows hosted on different machines. Users can serve their Flows, and others can access them by obtaining an instance of the served Flow and interacting with it through a proxy.\n",
    "\n",
    "In the following sections, we will guide you through:\n",
    "- Connecting to an existing colink server or starting a local one\n",
    "- Serving a Flow\n",
    "- Starting a worker to run Flows\n",
    "- Getting an instance of the served Flow via a proxy\n",
    "- Calling the Flow through the proxy"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.1 Connecting to an Existing Colink Server or Starting a Local One\n",
    "\n",
    "You have the option to either connect to an existing CoLink server or start a local one. Local servers are particularly useful for testing and development purposes. On the other hand, connecting to an existing server is beneficial when you need to access Flows served by others or serve your own Flows to others.\n",
    "\n",
    "For this AMLD workshop, we have initiated a CoLink server for you to connect to, allowing you to serve your Flows and access Flows served by others. However, for the purpose of this tutorial, we will guide you through starting a local server. Instructions for connecting to an existing server will be provided later during the workshop.\n",
    "\n",
    "To start a local server run the cell below (the server will run as long as your kernel is running)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Start the colink server and creates a user (you)\n",
    "# Returns the colink object who contains the information of the user and the server\n",
    "cl = colink_utils.start_colink_server()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.2 Serving A Flow\n",
    "\n",
    "To serve a Flow, you need to define its class and `run` method, along with its default configuration file (as we've done in the previous section for the `ReverseNumberAtomicFlow`). Once the Flow is defined, you can serve it using the `serve_flow` function. Essentially, this function saves the path to the Flow's class to the CoLink server's storage to be able to instantiate instances of it when requested.\n",
    "\n",
    "The key parameters of the `serve_flow` function are as follows:\n",
    "- `cl`: The CoLink object containing user and server information (this object is created when you connect to a server).\n",
    "- `flow_class_name`: Specifies the class of the Flow to be served (path to the class + class name).\n",
    "- `flow_endpoint`: The endpoint where the Flow will be served. This serves as a unique identifier for accessing the Flow. If the endpoint is already in use by another Flow, your Flow will not be served.\n",
    "- `singleton`: (Default is False) If set to True, only one instance of the Flow will be returned when a user requests it. In other words, multiple users can \"talk\" to the same Flow (e.g., all user \"talks\" to the same chatbot). If set to False, each user will get a different Flow instance (e.g., each user \"talks\"  to a different chatbots but of the same class).\n",
    "- `parallel_dispatch`: (Default is False) If set to True, the Flow can handle multiple requests simultaneously. If set to False, the Flow can only handle one request at a time. Note that this makes the Flow stateless.\n",
    "\n",
    "For example, in the cell below, we serve the `ReverseNumberAtomicFlow` at the endpoint \"reverse_number\" as a singleton.\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Serve the Flow\n",
    "serving.serve_flow(\n",
    "    cl=cl,\n",
    "    flow_class_name=\"ReverseNumberFlowModule.ReverseNumberAtomicFlow.ReverseNumberAtomicFlow\",\n",
    "    flow_endpoint=\"reverse_number\",\n",
    "    singleton=True,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.3 Starting a Worker to Run Flows\n",
    "\n",
    "As a server of the Flow, you must be the one to execute it when called. To execute a Flow, you need to start a worker. The worker is responsible for running the Flow's logic and will be activated whenever there are new messages to process. It's important to note that a single worker can handle multiple Flows; they are not limited to a single Flow exclusively.\n",
    "\n",
    "To start a worker, run the cell here below:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Start a worker thread to handle incoming messages\n",
    "run_dispatch_worker_thread(cl)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.4 Getting an Instance of the Served Flow via a Proxy\n",
    "\n",
    "Now that the Flow is served, you can access it by obtaining an instance of the served Flow through a proxy. The proxy is an `AtomicFlow` object that allows you to send messages to the requested Flow instance and receive responses from it. To obtain an instance of the served Flow, you can use the `get_flow_instance` function.\n",
    "\n",
    "The main parameters of the `get_flow_instance` function are:\n",
    "\n",
    "- `cl`: The CoLink object.\n",
    "- `flow_endpoint`: The endpoint where the Flow is served (this should match the endpoint used to serve the Flow).\n",
    "- `user_id`: The user ID of the user serving the Flow. Since we're serving a local server, you can simply set it to \"local\".\n",
    "- `config_overrides`: (Default is None) Any configuration overrides to be applied to the Flow. This is useful when you want to modify the default configuration of the Flow. For example, you might want to change the language of a chatbot. The configuration overrides are merged with the default configuration of the Flow, with the overrides taking precedence."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get an instance of the Flow\n",
    "proxy_reverse_number_flow = serving.get_flow_instance(\n",
    "    cl=cl,\n",
    "    flow_endpoint= \"reverse_number\", ### TODO - SPECIFY THE CORRECT FLOW ENDPOINT\n",
    "    user_id=\"local\",\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2.5. Calling the Reverse Number Flow via the Proxy\n",
    "\n",
    "Now that you have a proxy to the served Flow, you can send messages to it and receive responses. To send a message to the Flow, you need to package your data dictionary with the `package_input_message` method and then send a message via the proxy using the `get_reply_future` method. `get_reply_future` sends a message to the Flow with the data dictionary and the necessary information (in the `reply_data` attribute of the message) to get a reply back. It returns a future object that you can use to retrieve the reply from the Flow. The message can be retrieved from the future object using either the `get_data` method (which returns the `data` attribute of the reply message) or the `get_message` method. Both of these methods will block until the reply is received.\n",
    "\n",
    "Without further ado, let's send a message to the `ReverseNumberAtomicFlow` to reverse the number 12345!\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "input_data = {\"id\": 0, \"number\": 12345}\n",
    "\n",
    "# Package your data in a Flow Message\n",
    "\n",
    "## Option 1: Via the FlowMessage class\n",
    "# input_message = FlowMessage(\n",
    "#     data=input_data,\n",
    "# )\n",
    "\n",
    "## Option 2 (prefered): Via the package input message method\n",
    "input_message = proxy_reverse_number_flow.package_input_message(input_data)\n",
    "\n",
    "# Send a message to reverse number and ask to get an answer back in a future\n",
    "\n",
    "future = proxy_reverse_number_flow.get_reply_future(input_message)\n",
    "\n",
    "# Get the response from the future\n",
    "# To get the response as a data dictionary\n",
    "reply_data = future.get_data()\n",
    "# To get the response as a FlowMessage object\n",
    "reply_message = future.get_message()\n",
    "\n",
    "print(\"Data sent:\\n\",  input_data, \"\\n\")\n",
    "print(\"REPLY:\\n\", reply_data, \"\\n\")\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3.Writing your First CompositeFlow: ReverseNumberSequentialFlow\n",
    "\n",
    "In this section we will write a simple Composite Flow that reverses a number passed in the input message twice (e.g., 12345-> 54321 -> 12345).\n",
    "\n",
    "As mentioned in section 1, **Composite Flows:** are used to orchestrate other Flows in performing the computation. Just like **Atomic Flows**, Composite Flows have a `run` method that takes an input message as its input. The `run` method of a Composite Flow typically calls other Flows via their proxies, processes the responses, and sends a reply back. The Flows that a Composite Flow orchestrates are referred to as its subflows.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.1 Writing the ReverseNumberSequentialFlow Flow Class\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 3.1.1 ReverseNumberSequential"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%compile_and_writefile ReverseNumberFlowModule/ReverseNumberSequentialFlowBlocking.py\n",
    "\n",
    "\n",
    "from aiflows.base_flows import CompositeFlow\n",
    "from aiflows.messages import FlowMessage\n",
    "from aiflows.interfaces import KeyInterface\n",
    "class ReverseNumberSequentialFlowBlocking(CompositeFlow):\n",
    "    \n",
    "    def __init__(self, **kwargs):\n",
    "        super().__init__(**kwargs)\n",
    "        #~~~~~~~~~~~ Key Transformation solution 1 ~~~~~~~~~~~\n",
    "        self.input_interface_second_reverse_flow = KeyInterface(\n",
    "            keys_to_rename= {\"reversed_number\": \"number\"},\n",
    "            keys_to_select= [\"number\"],\n",
    "        )\n",
    "    \n",
    "        self.ouput_interface_reply = KeyInterface(\n",
    "            keys_to_rename= {\"reversed_number\": \"output_number\"},\n",
    "            keys_to_select = [\"output_number\"],\n",
    "        )\n",
    "                \n",
    "    # Customize the logic within this function as needed for your specific flow requirements.\n",
    "    def run(self, input_message: FlowMessage):\n",
    "        \n",
    "        # Call first reverse number\n",
    "        #### TODO send a message and get the future (see the previous cell for inspiration)\n",
    "        future_first_reverse = self.subflows[\"first_reverse_flow\"].get_reply_future(\n",
    "            input_message\n",
    "        )\n",
    "        # Get response from first reverse number\n",
    "         #### TODO send a message and get the future (see the previous cell for inspiration)\n",
    "        first_reverse_response = future_first_reverse.get_message()\n",
    "        \n",
    "        # Prepare response for the second reverse number\n",
    "        first_reverse_response = self.input_interface_second_reverse_flow(first_reverse_response) \n",
    "        \n",
    "        # Call second reverse number:\n",
    "        future_second_reverse = self.subflows[\"second_reverse_flow\"].get_reply_future(\n",
    "            first_reverse_response\n",
    "        )\n",
    "        \n",
    "        second_reverse_response = future_second_reverse.get_data()\n",
    "        \n",
    "        # Can call key interface tranformation on both message and dictionaries\n",
    "        # prepare response for the initial message\n",
    "        response = self.ouput_interface_reply(second_reverse_response)\n",
    "        \n",
    "        reply = self.package_output_message(\n",
    "            input_message = input_message,\n",
    "            response = response\n",
    "        )\n",
    "        \n",
    "        ### Send back reply\n",
    "        self.send_message(\n",
    "            reply\n",
    "        )\n",
    "        "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Curious about how `KeyInterface` works? The `KeyInterface` class is helpful when you need to apply transformations to the data dictionary when passing from one Flow to another and to pass a dictionary that satisfies the `input_interface` of the Flow you wish to call. It allows you to define these transformations in a structured way. While you can apply these transformations directly in the `run` method of the Composite Flow, the `KeyInterface` class provides a more organized approach.\n",
    "\n",
    "Check out the [Appendix](#4-appendix) for a detailed explanation of the `KeyInterface` class."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "default_config_reverse_number_sequential_blocking = \\\n",
    "{\n",
    "    \"name\": \"ReverseNumberTwiceBlocking\",\n",
    "    \"description\": \"A sequential Flow that reverses a number twice.\",\n",
    "\n",
    "    \"_target_\": \"ReverseNumberFlowModule.ReverseNumberSequentialFlowBlocking.ReverseNumberSequentialFlowBlocking.instantiate_from_default_config\", # TODO: Define the target  (see previous config fo inspiration)\n",
    "        \n",
    "    \"input_interface\": \"number\",\n",
    "    \"output_interface\": \"output_number\",\n",
    "    \n",
    "    \"subflows_config\": {\n",
    "        \"first_reverse_flow\": {\n",
    "            \"user_id\": \"local\",\n",
    "            \"flow_endpoint\": \"reverse_number\", ### TODO - SPECIFY THE CORRECT FLOW ENDPOINT (HINT: Endpoint of ReverseNumberAtomic) \n",
    "            \"name\": \"A proxy Flow that calls reverse number to reverse number AGAIN.\",\n",
    "            \"description\": \"A proxy Flow that calls reverse number to reverse number.\",\n",
    "        },\n",
    "        \"second_reverse_flow\": {\n",
    "            \"user_id\": \"local\",\n",
    "            \"flow_endpoint\": \"reverse_number\", ### TODO - SPECIFY THE CORRECT FLOW ENDPOINT (HINT: Endpoint of ReverseNumberAtomic) \n",
    "            \"name\": \"Proxy Second Reverse\",\n",
    "            \"description\": \"A proxy Flow that calls reverse number to reverse number AGAIN.\",\n",
    "        },\n",
    "    }\n",
    "}\n",
    "dict_to_yaml(default_config_reverse_number_sequential_blocking, \"ReverseNumberFlowModule/ReverseNumberSequentialFlowBlocking.yaml\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's dissect the new configuration file for the `ReverseNumberSequentialFlowBlocking`:\n",
    "\n",
    "- **`name` and `description`, `_target` , `input_interface`, `ouput_interface`**: These parameters are described in the previous section.\n",
    "\n",
    "- **`subflows_config`**:  Each subflow is articulated as a key-item pair within a dictionary. The key denotes the name assigned to the subflow, while the corresponding item is a dictionary encapsulating the configuration of the subflow. In this instance, subflows are outlined with their default configuration, incorporating overrides for the name and description of each Flow. Noteworthy keys:\n",
    "  - `flow_endpoint` and `user_id` match the `reverse_number` endpoint and `local` user ID, respectively, which were used to serve the `ReverseNumberAtomicFlow`. This allows us to get an instance of the Flow we served earlier when calling `get_flow_instance` (we don't have to serve `ReverseNumberAtomicFlow` again)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Serve the Flow (just like before)\n",
    "serving.serve_flow(\n",
    "    cl=cl,\n",
    "    flow_class_name= \"ReverseNumberFlowModule.ReverseNumberSequentialFlowBlocking.ReverseNumberSequentialFlowBlocking\", # TODO: Define the flow class name (Hint: see serving of ReverseNumberAtomic for inspiration)\n",
    "    flow_endpoint=\"reverse_sequential\",\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Start a second worker thread (since the Flow uses blocking calls, 2 workers are needed)\n",
    "run_dispatch_worker_thread(cl)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get an instance of the Flow and run it !!\n",
    "proxy_reverse_sequential_blocking = serving.get_flow_instance(\n",
    "    cl=cl,\n",
    "    flow_endpoint=\"reverse_sequential\",\n",
    "    user_id=\"local\",\n",
    "    config_overrides=default_config_reverse_number_sequential_blocking,\n",
    ")\n",
    "\n",
    "input_data = {\"id\": 0, \"number\": 12345}\n",
    "\n",
    "############ TODO: FILL OUT: (CHECK ReverseNumberAtomicForInspiration) ##########\n",
    "\n",
    "# Package your data in a Flow Message\n",
    "## Via the package input message method\n",
    "input_message = proxy_reverse_sequential_blocking.package_input_message(input_data) \n",
    "\n",
    "# Send a message to reverse number and ask to get an answer back in a future\n",
    "\n",
    "future = proxy_reverse_sequential_blocking.get_reply_future(input_message)\n",
    "\n",
    "# Get the response from the future\n",
    "# To get the response as a data dictionary\n",
    "reply_data = future.get_data()\n",
    "\n",
    "print(\"Data sent:\\n\",  input_data, \"\\n\")\n",
    "print(\"REPLY:\\n\", reply_data, \"\\n\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 3.1.2. ReverseNumberSequential with Non-Blocking Calls\n",
    "\n",
    "In the previous example, we used blocking calls to the subflows. In this example, we will use non-blocking calls. The description here below explains the differences between blocking and non-blocking calls.\n",
    "\n",
    "#### **Blocking vs Non-Blocking Calls**\n",
    "\n",
    "There are two ways to call a subflow from a Composite Flow via its proxy:\n",
    "1. **Blocking Calls**: The Composite Flow sends a message to the subflow and waits for the response before proceeding (and therefore stays in the `run` method until the response is received). Blocking calls are made using the `get_reply_future` method we introduced in the previous section (`get_reply_future` return a future object that you can use to retrieve the reply from the Flow via `future.get_data()` or `future.get_message()`, these \"get\" methods are what are blocking).\n",
    "2. **Non-Blocking Calls**: The Composite Flow sends a message to the subflow and proceeds without waiting for the response. The response is expected to be sent in the input message queue of the Composite Flow. Therefore, you must exit the `run` method and the Flow will be called again when the response is received. Non-blocking calls are made using the `send_message` method.\n",
    "\n",
    "##### Pros and cons of Blocking vs Non-Blocking Call? \n",
    "\n",
    "- **Blocking calls** might be more intuitive to use and understand, but they are less efficient because the Composite Flow is blocked while waiting for the response. Meaning that you are blocking one worker for the duration of the call. Therefore, you must have at least 2 worker threads running to be able to make a blocking call to a subflow and still be able to process other messages.\n",
    "\n",
    "- **Non-blocking calls** are more efficient because the Composite Flow is not blocked while waiting for the response (one worker can run the whole Flow). However, they are more complex to use because you must manage the state of the Flow and the responses.\n",
    "\n",
    "In the end, the choice between blocking and non-blocking calls depends on the specific use case, the requirements of the Flow and your preference.\n",
    "\n",
    "Here below we showcase the implementation of the `ReverseNumberSequentialFlow` class with both blocking and non-blocking calls to the `ReverseNumberAtomicFlow`.\n",
    "\n",
    "Note: Any data that you must save from one `run` method pass to another must be saved in the `self.flow_state` attribute of the Flow. This attribute is a dictionary that persists between calls to the `run` method. You can use it to save any data you need to keep between calls."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%compile_and_writefile ReverseNumberFlowModule/ReverseNumberSequentialFlowNonBlocking.py\n",
    "\n",
    "\n",
    "from aiflows.base_flows import CompositeFlow\n",
    "from aiflows.messages import FlowMessage\n",
    "from aiflows.interfaces import KeyInterface\n",
    "class ReverseNumberSequentialFlowNonBlocking(CompositeFlow):\n",
    "    \n",
    "    def __init__(self, **kwargs):\n",
    "        super().__init__(**kwargs)\n",
    "        #~~~~~~~~~~~ Key Transformation solution 1 ~~~~~~~~~~~\n",
    "        self.input_interface_second_reverse_flow = KeyInterface(\n",
    "            keys_to_rename= {\"reversed_number\": \"number\"},\n",
    "            keys_to_select= [\"number\"],\n",
    "        )\n",
    "    \n",
    "        self.ouput_interface_reply = KeyInterface(\n",
    "            keys_to_rename= {\"reversed_number\": \"output_number\"},\n",
    "            keys_to_select = [\"output_number\"],\n",
    "        )\n",
    "        \n",
    "        self.next_state_dict = {\n",
    "            \"first_reverse_flow\": \"second_reverse_flow\",\n",
    "            \"second_reverse_flow\": \"reply_to_message\",\n",
    "            \"reply_to_message\": \"first_reverse_flow\"\n",
    "        }\n",
    "    def set_up_flow_state(self):\n",
    "        super().set_up_flow_state()\n",
    "        self.flow_state = {\"current_state\": \"first_reverse_flow\"}\n",
    "        \n",
    "    def get_next_state(self):\n",
    "        return self.next_state_dict[self.flow_state[\"current_state\"]]\n",
    "        \n",
    "    # Customize the logic within this function as needed for your specific Flow requirements.\n",
    "    def run(self, input_message: FlowMessage):\n",
    "        \n",
    "        # Run here is a bit like a switch statement where we decide which flow to call next.\n",
    "        # We then call the next Flow and pass the input message to it. which we expect to get a reply from \n",
    "        # back in the input queue (which will call the run method againg)\n",
    "        current_state = self.flow_state[\"current_state\"]\n",
    "        \n",
    "        # Case where we need to reverse the number for the first time\n",
    "        if current_state == \"first_reverse_flow\":\n",
    "            #Save the initial message to the state\n",
    "            self.flow_state[\"initial_message\"] = input_message\n",
    "            \n",
    "            # Prepares message by editing `reply_data` so that it can get the reply from the first Flow\n",
    "            # back in the input queue of the composite flow \n",
    "            message_for_first_reverse_flow = self.package_input_message(input_message.data)\n",
    "            # Call the first Flow with non blocking call\n",
    "            self.subflows[\"first_reverse_flow\"].get_reply(\n",
    "                message_for_first_reverse_flow\n",
    "            )\n",
    "        \n",
    "        #Case where we need to reverse the number for the second time\n",
    "        elif current_state == \"second_reverse_flow\":\n",
    "            \n",
    "            # Applies a transformation to the input message (renames keys of dictonary so that they match the\n",
    "            # required format of the second flow)\n",
    "            message = self.input_interface_second_reverse_flow(input_message)\n",
    "            \n",
    "            # Prepares message by editing `reply_data` so that it can get the reply from the first Flow\n",
    "            # back in the input queue of the composite flow\n",
    "            \n",
    "            ########## TODO: PACKAGE MESSAGE AS INPUT MESSAGE FOR SECOND FLOW AND MAKE A NON-BLOCKING-CALL##########\n",
    "            message_for_second_reverse_flow = self.package_input_message(input_message.data)\n",
    "            # call the first Flow with non blocking call\n",
    "            self.subflows[\"second_reverse_flow\"].get_reply(\n",
    "                message_for_second_reverse_flow\n",
    "            )\n",
    "        \n",
    "        # Case where we need to reply to the initial message (we've already reversed the number twice)\n",
    "        else:\n",
    "            message = self.ouput_interface_reply(input_message)\n",
    "            \n",
    "            # package ouput message to send back\n",
    "                # This method packages `response` in a FlowMessage object \n",
    "                # containing the necessary metadata to send the message back\n",
    "                # to the sender of the input message. \n",
    "            reply = self.package_output_message(\n",
    "                input_message = self.flow_state[\"initial_message\"],\n",
    "                response = message.data\n",
    "            )\n",
    "            #send back the reply to initial caller of the Flow\n",
    "            self.send_message(reply)\n",
    "            \n",
    "        self.flow_state[\"current_state\"] = self.get_next_state()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The config is identical to the one above (except for `_target_` ) so we won't go through it again."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "default_config_reverse_number_sequential_nonblocking = \\\n",
    "{\n",
    "    \"name\": \"ReverseNumberTwice\",\n",
    "    \"description\": \"A sequential Flow that reverses a number twice.\",\n",
    "\n",
    "    # TODO: Define the target\n",
    "    \"_target_\": \\\n",
    "        \"ReverseNumberFlowModule.ReverseNumberSequentialFlowNonBlocking.ReverseNumberSequentialFlowNonBlocking.instantiate_from_default_config\",\n",
    "\n",
    "    \"input_interface\": \"number\",\n",
    "    \"output_interface\": \"output_number\",\n",
    "    \n",
    "    \"subflows_config\": {\n",
    "        \"first_reverse_flow\": {\n",
    "            \"user_id\": \"local\",\n",
    "            \"flow_endpoint\": \"reverse_number\",\n",
    "            \"name\": \"A proxy Flow that calls reverse number to reverse number AGAIN.\",\n",
    "            \"description\": \"A proxy Flow that calls reverse number to reverse number.\",\n",
    "        },\n",
    "        \"second_reverse_flow\": {\n",
    "            \"user_id\": \"local\",\n",
    "            \"flow_endpoint\": \"reverse_number\",\n",
    "            \"name\": \"Proxy Second Reverse\",\n",
    "            \"description\": \"A proxy Flow that calls reverse number to reverse number AGAIN.\",\n",
    "        },\n",
    "    }\n",
    "}\n",
    "dict_to_yaml(default_config_reverse_number_sequential_nonblocking, \"ReverseNumberFlowModule/ReverseNumberSequentialFlowNonBlocking.yaml\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Serve the Flow (just like before)\n",
    "serving.serve_flow(\n",
    "    cl=cl,\n",
    "    flow_class_name=\"ReverseNumberFlowModule.ReverseNumberSequentialFlowNonBlocking.ReverseNumberSequentialFlowNonBlocking\",\n",
    "    flow_endpoint=\"ReverseNumberSequentialFlowNonBlocking\",\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "# Get an instance of the Flow and run it !!\n",
    "######### TODO: GET AN INSTANCE OF THE FLOW ########\n",
    "# HINT: Checkout ReverseNumberSequentialFlowNonBlocking for inspiration \n",
    "proxy_reverse_sequential_nonblocking = serving.get_flow_instance(\n",
    "    cl=cl,\n",
    "    flow_endpoint=\"ReverseNumberSequentialFlowNonBlocking\",\n",
    "    user_id=\"local\",\n",
    "    config_overrides=default_config_reverse_number_sequential_nonblocking,\n",
    ")\n",
    "input_data = {\"id\": 0, \"number\": 12345}\n",
    "\n",
    "# Package your data in a Flow Message\n",
    "## Via the package input message method\n",
    "input_message = proxy_reverse_sequential_nonblocking.package_input_message(input_data)\n",
    "\n",
    "# Send a message to reverse number and ask to get an answer back in a future\n",
    "\n",
    "future = proxy_reverse_sequential_nonblocking.get_reply_future(input_message)\n",
    "\n",
    "# Get the response from the future\n",
    "# To get the response as a data dictionary\n",
    "reply_data = future.get_data()\n",
    "# To get the response as a FlowMessage object\n",
    "reply_message = future.get_message()\n",
    "\n",
    "print(\"Data sent:\\n\",  input_data, \"\\n\")\n",
    "print(\"REPLY:\\n\", reply_data, \"\\n\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4. Appendix"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### **Key Interfaces**\n",
    "\n",
    "The `KeyInterface` class is helpful when you need to apply transformations to the data dictionary when passing from one Flow to another and to pass a dictionary that satisfies the `input_interface` of the Flow you wish to call. It allows you to define these transformations in a structured way. While you can apply these transformations directly in the `run` method of the Composite Flow, the `KeyInterface` class provides a more organized approach. The `KeyInterface` class takes the following input parameters (which define the transformations) and executes them in the order they are defined:\n",
    "- `keys_to_rename`: A dictionary where the keys are the keys to rename and the values are the new names of the keys. For example, `{\"old_key\": \"new_key\"}` transforms `{\"old_key\": 1}` to `{\"new_key\": 1}`.\n",
    "\n",
    "- `keys_to_copy`: A dictionary where the keys are the keys to copy and the values are the new names of the keys. For example, `{\"key_a\": \"key_b\"}` transforms `{\"key_a\": 1}` to `{\"key_a\": 1, \"key_b\": 1}`.\n",
    "- `keys_to_set`: A dictionary where the keys are the keys to set and the values are the values to set. For example, `{\"key_a\": 1}` transforms `{}` to `{\"key_a\": 1}`.\n",
    "\n",
    "- `additional_transformations`: A list of Callables where you can define custom transformations. Each Callable should take the arguments `data_dict` and return a dictionary as output.For example:\n",
    "    ```python\n",
    "        def increment_key_a(data_dict, **kwargs): \n",
    "            return {**input_data,**{\"key_a\": input_data[\"key_a\"] + 1}}\n",
    "    ```\n",
    "\n",
    "- `key_to_select`: A list of keys to select from the dictionary. For example, `[\"key_a\"]` transforms `{\"key_a\": 1, \"key_b\": 2}` to `{\"key_a\": 1}`.\n",
    "\n",
    "- `key_to_delete`: A list of keys to delete from the dictionary. For example, `[\"key_a\"]` transforms `{\"key_a\": 1, \"key_b\": 2}` to `{\"key_b\": 2}`.\n",
    "\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "coflows",
   "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.11.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}