<jupyter_start><jupyter_text>Leveraging AsyncIO in inference pipelinesTempo includes experimental support for `asyncio`, which provides a way to optimise pipelines.In particular, `asyncio` can be beneficial in scenarios where most of the heavy lifting is done by downstream models and the pipeline just orchestrates calls across these models.In this case, most of the time within the pipeline will be spent waiting for the requests from downstream models to come back.`asyncio` will allow us to process other incoming requests during this waiting time.This example will walk us through the process of setting up an asynchronous pipeline.As you will see, it's quite similar to the usual synchronous pipelines.  PrerequisitesThis notebooks needs to be run in the `tempo-examples` conda environment defined below. Create from project root folder:```bashconda env create --name tempo-examples --file conda/tempo-examples.yaml```  Project Structure<jupyter_code>!tree -P "*.py"  -I "__init__.py|__pycache__" -L 2<jupyter_output>[01;34m.[00m
├── [01;34martifacts[00m
│   ├── [01;34mclassifier[00m
│   ├── [01;34msklearn[00m
│   └── [01;34mxgboost[00m
└── [01;34msrc[00m
    ├── constants.py
    ├── data.py
    ├── tempo.py
    └── train.py

5 directories, 4 files<jupyter_text>Train ModelsThis section is where as a data scientist you do your work of training models and creating artfacts.For this example, we will train two sklearn and xgboost classification models using the iris dataset.These models will be used by our inference pipeline.<jupyter_code>import logging
from tempo.utils import logger

logger.setLevel(logging.ERROR)
logging.basicConfig(level=logging.ERROR)
# %load src/train.py
import os

import joblib
from sklearn.linear_model import LogisticRegression
from src.constants import SKLearnFolder, XGBoostFolder
from src.data import IrisData
from xgboost import XGBClassifier


def train_sklearn(data: IrisData):
    logreg = LogisticRegression(C=1e5)
    logreg.fit(data.X, data.y)

    model_path = os.path.join(SKLearnFolder, "model.joblib")
    with open(model_path, "wb") as f:
        joblib.dump(logreg, f)


def train_xgboost(data: IrisData):
    clf = XGBClassifier()
    clf.fit(data.X, data.y)

    model_path = os.path.join(XGBoostFolder, "model.json")
    clf.save_model(model_path)

from src.data import IrisData
from src.train import train_sklearn, train_xgboost

data = IrisData()

train_sklearn(data)
train_xgboost(data)<jupyter_output>[18:05:52] WARNING: ../src/learner.cc:1095: Starting in XGBoost 1.3.0, the default evaluation metric used with the objective 'multi:softprob' was changed from 'merror' to 'mlogloss'. Explicitly set eval_metric if you'd like to restore the old behavior.<jupyter_text>Create Tempo ArtifactsHere we create the Tempo models and orchestration Pipeline for our final service using our models.For illustration the final service will call the sklearn model and based on the result will decide to return that prediction or call the xgboost model and return that prediction instead.<jupyter_code>from src.tempo import classifier
# %load src/tempo.py
import numpy as np
from src.constants import ClassifierFolder, SKLearnFolder, XGBoostFolder

from tempo import ModelFramework, PipelineModels
from tempo.aio import Model, pipeline

SKLearnModel = Model(
    name="test-iris-sklearn",
    platform=ModelFramework.SKLearn,
    local_folder=SKLearnFolder,
    uri="s3://tempo/basic/sklearn",
    description="An SKLearn Iris classification model",
)

XGBoostModel = Model(
    name="test-iris-xgboost",
    platform=ModelFramework.XGBoost,
    local_folder=XGBoostFolder,
    uri="s3://tempo/basic/xgboost",
    description="An XGBoost Iris classification model",
)


@pipeline(
    name="classifier",
    models=PipelineModels(sklearn=SKLearnModel, xgboost=XGBoostModel),
    local_folder=ClassifierFolder,
)
async def classifier(payload: np.ndarray) -> np.ndarray:
    res1 = await classifier.models.sklearn(input=payload)
    if res1[0] > 0.7:
        return res1

    return await classifier.models.xgboost(input=payload)<jupyter_output><empty_output><jupyter_text>Save Classifier EnvironmentIn preparation for running our models we save the Python environment needed for the orchestration to run as defined by a `conda.yaml` in our project.<jupyter_code>!ls artifacts/classifier/conda.yaml
import tempo

tempo.save(classifier)<jupyter_output>Collecting packages...
Packing environment at '/home/clive/anaconda3/envs/tempo-330c15d8-a189-45a6-abc3-a27f39b6a7c5' to '/home/clive/work/mlops/fork-tempo/docs/examples/asyncio/artifacts/classifier/environment.tar.gz'
[########################################] | 100% Completed | 11.2s<jupyter_text>Test Locally on DockerHere we test our models using production images but running locally on Docker. This allows us to ensure the final production deployed model will behave as expected when deployed.<jupyter_code>from tempo.aio import deploy_local
remote_model = deploy_local(classifier)
import numpy as np
await remote_model.predict(np.array([[1, 2, 3, 4]]))
print(await remote_model.predict(np.array([[0, 0, 0,0]])))
print(await remote_model.predict(np.array([[5.964,4.006,2.081,1.031]])))
remote_model.undeploy()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.09798708697303456, 'fraction_numerical': 0.021458412457273073, 'mean_word_length': 5.292712066905615, 'pattern_counts': {'":': 0, '<': 18, '<?xml version=': 0, '>': 20, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '137611', 'license': 'Apache-2.0', 'n_tokens_mistral': 1754, 'n_tokens_neox': 1630, 'n_words': 510, 'path': 'docs/examples/asyncio/README.ipynb', 'repo_name': 'sakoush/tempo', 'size': '5266'}