Workflow executor example workflow API

WorkflowExecAgent/README.md

@@ -4,13 +4,25 @@
 
 GenAI Workflow Executor Example showcases the capability to handle data/AI workflow operations via LangChain agents to execute custom-defined workflow-based tools. These workflow tools can be interfaced from any 3rd-party tools in the market (no-code/low-code/IDE) such as Alteryx, RapidMiner, Power BI, Intel Data Insight Automation which allows users to create complex data/AI workflow operations for different use-cases.
 
+### Definitions
+
+Before we begin, here are the definitions to some terms for clarity:
+
+- servable/serving workflow - A workflow made ready to be executed through API. It should be able to accept parameter injection for workflow scheduling and have a way to retrieve the final output data. It should also have a unique workflow ID for referencing. For platform providers guide to create their own servable workflows compatible with this example, refer to [Workflow Building Platform](#workflow-building-platform)
+
+- SDK Class - Performs requests to interface with a 3rd-party API to perform workflow operations on the servable workflow. Found in `tools/sdk.py`.
+
+- workflow ID - A unique ID for the servable workflow.
+
+- workflow instance - An instance created from the servable workflow. It is represented as a `Workflow` class created using `DataInsightAutomationSDK.create_workflow()` under `tools/sdk.py`. Contains methods to `start`, `get_status` and `get_results` from the workflow.
+
 ### Workflow Executor
 
-This example demonstrates a single React-LangGraph with a `Workflow Executor` tool to ingest a user prompt to execute workflows and return an agent reasoning response based on the workflow output data.
+Strategy - This example demonstrates a single React-LangGraph with a `Workflow Executor` tool to ingest a user prompt to execute workflows and return an agent reasoning response based on the workflow output data.
 
 First the LLM extracts the relevant information from the user query based on the schema of the tool in `tools/tools.yaml`. Then the agent sends this `AgentState` to the `Workflow Executor` tool.
 
-`Workflow Executor` tool uses `EasyDataSDK` class as seen under `tools/sdk.py` to interface with several high-level API's. There are 3 steps to this tool implementation:
+`Workflow Executor` tool requires a SDK class to call the servable workflow API. In the code, `DataInsightAutomationSDK` is the example class as seen under `tools/sdk.py` to interface with several high-level API's. There are 3 steps to this tool implementation:
 
 1. Starts the workflow with workflow parameters and workflow id extracted from the user query.
 
@@ -26,37 +38,50 @@ Below shows an illustration of this flow:
 
 ### Workflow Serving for Agent
 
+#### Workflow Building Platform
+
+The first step is to prepare a servable workflow using a platform with the capabilities to do so.
+
 As an example, here we have a Churn Prediction use-case workflow as the serving workflow for the agent execution. It is created through Intel Data Insight Automation platform. The image below shows a snapshot of the Churn Prediction workflow.
 
 ![image](https://github.com/user-attachments/assets/c067f8b3-86cf-4abc-a8bd-51a98de8172d)
 
-The workflow contains 2 paths which can be seen in the workflow illustrated, the top path and bottom path.
+The workflow contains 2 paths which can be seen in the workflow illustrated, the top and bottom paths.
 
-1. Top path - The training path which ends at the random forest classifier node is the training path. The data is cleaned through a series of nodes and used to train a random forest model for prediction.
+1. Top path (Training path) - Ends at the random forest classifier node is the training path. The data is cleaned through a series of nodes and used to train a random forest model for prediction.
 
-2. Bottom path - The inference path where trained random forest model is used for inferencing based on input parameter.
+2. Bottom path (Inference path) - where trained random forest model is used for inferencing based on input parameter.
 
 For this agent workflow execution, the inferencing path is executed to yield the final output result of the `Model Predictor` node. The same output is returned to the `Workflow Executor` tool through the `Langchain API Serving` node.
 
-There are `Serving Parameters` in the workflow, which are the tool input variables used to start a workflow instance obtained from `params` the LLM extracts from the user query. Below shows the parameter configuration option for the Intel Data Insight Automation workflow UI.
+There are `Serving Parameters` in the workflow, which are the tool input variables used to start a workflow instance at runtime obtained from `params` the LLM extracts from the user query. Below shows the parameter configuration option for the Intel Data Insight Automation workflow UI.
 
-![image](https://github.com/user-attachments/assets/ce8ef01a-56ff-4278-b84d-b6e4592b28c6)
+<img src="https://github.com/user-attachments/assets/ce8ef01a-56ff-4278-b84d-b6e4592b28c6" alt="image" width="500"/>
 
 Manually running the workflow yields the tabular data output as shown below:
 
 ![image](https://github.com/user-attachments/assets/241c1aba-2a24-48da-8005-ec7bfe657179)
 
 In the workflow serving for agent, this output will be returned to the `Workflow Executor` tool. The LLM can then answer the user's original question based on this output.
 
-To start prompting the agent microservice, we will use the following command for this use case:
+When the workflow is configured as desired, transform this into a servable workflow. We turn this workflow into a servable workflow format so that it can be called through API to perform operations on it. Data Insight Automation has tools to do this for its own workflows.
+
+> [!NOTE]
+> Remember to create a unique workflow ID along with the servable workflow.
+
+#### Using Servable Workflow
+
+Once we have our servable workflow ready, the serving workflow API can be prepared to accept requests from the SDK class. Refer to [Start Agent Microservice](#start-agent-microservice) on how to do this.
+
+To start prompting the agent microservice, we will use the following command for this churn prediction use-case:
 
 ```sh
 curl http://${ip_address}:9090/v1/chat/completions -X POST -H "Content-Type: application/json" -d '{
      "query": "I have a data with gender Female, tenure 55, MonthlyAvgCharges 103.7. Predict if this entry will churn. My workflow id is '${workflow_id}'."
     }'
 ```
 
-The user has to provide a `workflow_id` and workflow `params` in the query. `workflow_id` a unique id used for serving the workflow to the microservice. Notice that the `query` string includes all the workflow `params` which the user has defined in the workflow. The LLM will extract these parameters into a dictionary format for the workflow `Serving Parameters` as shown below:
+The user has to provide a `workflow_id` and workflow `params` in the query. Notice that the `query` string includes all the workflow `params` which the user has defined in the workflow. The LLM will extract these parameters into a dictionary format for the workflow `Serving Parameters` as shown below:
 
 ```python
 params = {"gender": "Female", "tenure": 55, "MonthlyAvgCharges": 103.7}
@@ -72,6 +97,16 @@ And finally here are the results from the microservice logs:
 
 ### Start Agent Microservice
 
+For an out-of-box experience there is an example workflow serving API service prepared for users under `tests/example_workflow` to interface with the SDK. This section will guide users on setting up this service as well. Users may modify the logic, add your own database etc for your own use-case.
+
+There are 3 services needed for the setup:
+
+1. Agent microservice
+
+2. LLM inference service - specified as `llm_endpoint_url`.
+
+3. workflow serving API service - specified as `SDK_BASE_URL`
+
 Workflow Executor will have a single docker image. First, build the agent docker image.
 
 ```sh
@@ -83,8 +118,9 @@ docker compose -f build.yaml build --no-cache
 Configure `GenAIExamples/WorkflowExecAgent/docker_compose/.env` file with the following. Replace the variables according to your usecase.
 
 ```sh
-export SDK_BASE_URL=${SDK_BASE_URL}
-export SERVING_TOKEN=${SERVING_TOKEN}
+export wf_api_port=5000     # workflow serving API port to use
+export SDK_BASE_URL=http://$(hostname -I | awk '{print $1}'):${wf_api_port}/      # The workflow server will use this example workflow API url
+export SERVING_TOKEN=${SERVING_TOKEN}           # For example_workflow, can be empty
 export HUGGINGFACEHUB_API_TOKEN=${HF_TOKEN}
 export llm_engine=${llm_engine}
 export llm_endpoint_url=${llm_endpoint_url}
@@ -106,9 +142,18 @@ cd $WORKDIR/GenAIExamples/WorkflowExecAgent/docker_compose
 docker compose -f compose.yaml up -d
 ```
 
+To launch the example workflow API server, open a new terminal and run the following:
+
+```sh
+cd $WORKDIR/GenAIExamples/WorkflowExecAgent/tests/example_workflow
+. launch_workflow_service.sh
+```
+
+`launch_workflow_service.sh` will setup all the packages locally and launch the uvicorn server to host the API on port 5000. For a Dockerfile method, please refer to `Dockerfile.example_workflow_api` file.
+
 ### Validate service
 
-The microservice logs can be viewed using:
+The agent microservice logs can be viewed using:
 
 ```sh
 docker logs workflowexec-agent-endpoint
@@ -120,7 +165,7 @@ You can validate the service using the following command:
 
 ```sh
 curl http://${ip_address}:9090/v1/chat/completions -X POST -H "Content-Type: application/json" -d '{
-     "query": "I have a data with gender Female, tenure 55, MonthlyAvgCharges 103.7. Predict if this entry will churn. My workflow id is '${workflow_id}'."
+     "query": "I have a data with gender Female, tenure 55, MonthlyCharges 103.7, TotalCharges 1840.75. Predict if this entry will churn. My workflow id is '${workflow_id}'."
     }'
 ```
 

WorkflowExecAgent/tests/3_launch_agent_service.sh

@@ -0,0 +1,38 @@
+#!/bin/bash
+# Copyright (C) 2024 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+set -e
+
+WORKPATH=$(dirname "$PWD")
+vllm_port=${vllm_port}
+[[ -z "$vllm_port" ]] && vllm_port=8084
+export WORKDIR=$WORKPATH/../../
+echo "WORKDIR=${WORKDIR}"
+export SDK_BASE_URL=$1
+export SERVING_TOKEN=${SERVING_TOKEN}
+export HF_TOKEN=${HUGGINGFACEHUB_API_TOKEN}
+export llm_engine=vllm
+export ip_address=$(hostname -I | awk '{print $1}')
+export llm_endpoint_url=http://${ip_address}:${vllm_port}
+export model=mistralai/Mistral-7B-Instruct-v0.3
+export recursion_limit=25
+export temperature=0
+export max_new_tokens=1000
+export TOOLSET_PATH=$WORKDIR/GenAIExamples/WorkflowExecAgent/tools/
+
+function start_agent() {
+    echo "Starting Agent services"
+    cd $WORKDIR/GenAIExamples/WorkflowExecAgent/docker_compose/intel/cpu/xeon
+    WORKDIR=$WORKPATH/docker_image_build/ docker compose -f compose_vllm.yaml up -d
+    echo "Waiting agent service ready"
+    sleep 5s
+}
+
+function main() {
+    echo "==================== Start agent service ===================="
+    start_agent
+    echo "==================== Agent service started ===================="
+}
+
+main

WorkflowExecAgent/tests/3_launch_example_wf_api.sh

@@ -0,0 +1,40 @@
+#!/bin/bash
+# Copyright (C) 2024 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+set -e
+
+wf_api_port=${wf_api_port}
+[[ -z "$wf_api_port" ]] && wf_api_port=5000
+WORKPATH=$(dirname "$PWD")
+LOG_PATH="$WORKPATH/tests/example_workflow"
+export WORKDIR=$WORKPATH/../../
+echo "WORKDIR=${WORKDIR}"
+
+function start_example_workflow_api() {
+    echo "Starting example workflow API"
+    cd $WORKDIR/GenAIExamples/WorkflowExecAgent/tests/example_workflow
+    docker build -f Dockerfile.example_workflow_api -t example-workflow-service .
+    docker run -d -p ${wf_api_port}:${wf_api_port} --rm --network=host --name example-workflow-service -it example-workflow-service
+    echo "Waiting example workflow API ready"
+    until [[ "$n" -ge 100 ]] || [[ $ready == true ]]; do
+        docker logs example-workflow-service &> ${LOG_PATH}/example-workflow-service.log
+        n=$((n+1))
+        if grep -q "Uvicorn running on" ${LOG_PATH}/example-workflow-service.log; then
+            break
+        fi
+        if grep -q "No such container" ${LOG_PATH}/example-workflow-service.log; then
+            echo "container example-workflow-service not found"
+            exit 1
+        fi
+        sleep 5s
+    done
+}
+
+function main() {
+    echo "==================== Start example workflow API ===================="
+    start_example_workflow_api
+    echo "==================== Example workflow API started ===================="
+}
+
+main

WorkflowExecAgent/tests/3_launch_and_validate_agent.sh → WorkflowExecAgent/tests/4_validate_agent.sh

@@ -5,30 +5,11 @@
 set -e
 
 WORKPATH=$(dirname "$PWD")
-workflow_id=9809
-vllm_port=${vllm_port}
-[[ -z "$vllm_port" ]] && vllm_port=8084
 export WORKDIR=$WORKPATH/../../
 echo "WORKDIR=${WORKDIR}"
-export SDK_BASE_URL=${SDK_BASE_URL}
-export SERVING_TOKEN=${SERVING_TOKEN}
-export HF_TOKEN=${HUGGINGFACEHUB_API_TOKEN}
-export llm_engine=vllm
 export ip_address=$(hostname -I | awk '{print $1}')
-export llm_endpoint_url=http://${ip_address}:${vllm_port}
-export model=mistralai/Mistral-7B-Instruct-v0.3
-export recursion_limit=25
-export temperature=0
-export max_new_tokens=1000
-export TOOLSET_PATH=$WORKDIR/GenAIExamples/WorkflowExecAgent/tools/
-
-function start_agent_and_api_server() {
-    echo "Starting Agent services"
-    cd $WORKDIR/GenAIExamples/WorkflowExecAgent/docker_compose/intel/cpu/xeon
-    WORKDIR=$WORKPATH/docker_image_build/ docker compose -f compose_vllm.yaml up -d
-    echo "Waiting agent service ready"
-    sleep 5s
-}
+query=$1
+validate_result=$2
 
 function validate() {
     local CONTENT="$1"
@@ -47,17 +28,13 @@ function validate() {
 function validate_agent_service() {
     echo "----------------Test agent ----------------"
     local CONTENT=$(curl http://${ip_address}:9090/v1/chat/completions -X POST -H "Content-Type: application/json" -d '{
-     "query": "I have a data with gender Female, tenure 55, MonthlyAvgCharges 103.7. Predict if this entry will churn. My workflow id is '${workflow_id}'."
+     "query": "'"${query}"'"
     }')
-    validate "$CONTENT" "The entry is not likely to churn" "workflowexec-agent-endpoint"
+    validate "$CONTENT" "$validate_result" "workflowexec-agent-endpoint"
     docker logs workflowexec-agent-endpoint
 }
 
 function main() {
-    echo "==================== Start agent ===================="
-    start_agent_and_api_server
-    echo "==================== Agent started ===================="
-
     echo "==================== Validate agent service ===================="
     validate_agent_service
     echo "==================== Agent service validated ===================="

WorkflowExecAgent/tests/example_workflow/Dockerfile.example_workflow_api

@@ -0,0 +1,11 @@
+FROM ubuntu:22.04
+
+RUN apt-get -qq update
+
+WORKDIR /home/ubuntu
+
+COPY launch_workflow_service.sh requirements.txt main.py workflow.py ./
+
+RUN chmod +x ./launch_workflow_service.sh
+
+CMD ["./launch_workflow_service.sh"]

WorkflowExecAgent/tests/example_workflow/launch_workflow_service.sh

@@ -0,0 +1,18 @@
+#!/bin/bash
+
+# Copyright (C) 2024 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+apt-get -qq -y install --no-install-recommends unzip curl ca-certificates
+apt-get -qq -y install --no-install-recommends python3 python3-pip
+
+curl -L -o ./archive.zip https://www.kaggle.com/api/v1/datasets/download/blastchar/telco-customer-churn
+unzip ./archive.zip -d ./
+rm ./archive.zip
+
+pip install virtualenv && \
+    virtualenv venv && \
+    source venv/bin/activate && \
+    pip install -r requirements.txt
+
+uvicorn main:app --reload --port=5000 --host=0.0.0.0

WorkflowExecAgent/tests/example_workflow/main.py

@@ -0,0 +1,58 @@
+# Copyright (C) 2024 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+
+import logging
+
+from fastapi import APIRouter, FastAPI
+from workflow import run_workflow
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.DEBUG)
+
+app = FastAPI()
+
+router = APIRouter(prefix="/serving", tags=["Workflow Serving"])
+
+app.results = {}
+
+
+@router.post("/servable_workflows/{wf_id}/start", summary="Start Workflow")
+async def start_workflow(wf_id: int, params: dict):
+    try:
+        app.results = run_workflow(params["params"])
+        wf_key = "example_key"
+        return {"msg": "ok", "wf_key": wf_key}
+
+    except Exception as e:
+        logging.error(e, exc_info=True)
+        return {"msg": "error occurred"}
+
+
+@router.get("/serving_workflows/{wf_key}/status", summary="Get Workflow Status")
+async def get_status(wf_key: str):
+    try:
+        if app.results:
+            status = "finished"
+        else:
+            status = "failed"
+
+        return {"workflow_status": status}
+    except Exception as e:
+        logging.error(e)
+        return {"msg": "error occurred"}
+
+
+@router.get("/serving_workflows/{wf_key}/results", summary="Get Workflow Results")
+async def get_results(wf_key: str):
+    try:
+        if app.results:
+            return app.results
+        else:
+            return {"msg": "There is an issue while getting results !!"}
+
+    except Exception as e:
+        logging.error(e)
+        return {"msg": "There is an issue while getting results !!"}
+
+
+app.include_router(router)

WorkflowExecAgent/tests/example_workflow/requirements.txt

@@ -0,0 +1,7 @@
+fastapi
+numpy
+pandas
+requests==2.28.1
+scikit-learn
+uvicorn
+xgboost