A Deep Dive Into BentoML

Remove obstacles to productionize your ML model using BentoML

Photo by Fatos Bytyqi on Unsplash

Recently, I’ve started deep-diving into BentoML for machine learning production use cases. And TBH, it has been the easiest journey to productionize a machine learning model, serving not one but multiple models with simple APIs.

Today, I will walk you through multiple examples of BentoML and ensure you’ll follow along and develop the same for your project.

For Non-Members: Read here!

For video tutorial: Getting Started With BentoML, Adaptive Batching with BentoML, and Model Composition In BentoML!

Why BentoML?

BentoML >> Flask, FastAPI, or raw Docker: for machine learning model deployment. It enables a quick transition from development to production. It streamlines the entire process: packaging, serving, and deployment into a unified, high-performance workflow.

Google Gemini

Environment Set Up

python3 -m venv bentoml-env
source bentoml-env/bin/activate
pip install bentoml torch transformers

Example One

Create a bentoml service for a machine learning model, and serve it as an API locally.

#bentoml_service.py
from __future__ import annotations
import bentoml

with bentoml.importing():
    from transformers import pipeline

EXAMPLE_INPUT = """Breaking News: In an astonishing turn of events, the small 
                town of Willow Creek has been taken by storm as local resident 
                Jerry Thompson's cat, Whiskers, performed what witnesses are 
                calling a 'miraculous and gravity-defying leap.' Eyewitnesses 
                report that Whiskers, an otherwise unremarkable tabby cat, 
                jumped a record-breaking 20 feet into the air to catch a fly. 
                The event, which took place in Thompson's backyard, is now 
                being investigated by scientists for potential breaches in 
                the laws of physics. Local authorities are considering a 
                town festival to celebrate what is being hailed as 
                'The Leap of the Century."""

@bentoml.service
class Summarization:
    def __init__(self) -> None:
        self.pipeline = pipeline('summarization')

    @bentoml.api
    def summarize(self, text: str = EXAMPLE_INPUT) -> str:
        result = self.pipeline(text)
        return f"Hello world! Here's your summary: {result[0]['summary_text']}"

In the **Summarization class**, the BentoML Service retrieves a pre-trained model and initializes a pipeline for text summarization.

**summarize function**: Serves as the API endpoint.

• Input as a string(Sample provided)
• Processes it through the pipeline
• Returns the summarized text.

In BentoML, a Service is a deployable and scalable unit.

• A Python class is turned into a BentoML service using the @bentoml.service decorator.
• It can manage states and their lifecycle, and expose one or multiple APIs accessible through HTTP.
• Each API within the Service is defined using the **@bentoml.api** decorator, specifying it as a Python function.

The **bentoml.importing()** is a context manager used for handling the import statements for dependencies required during serving, but may not be available in other situations.

Run BentoML Service

Once the service is up and running from the command below, we can see **localhost:3000** to view the swagger ui, and hit the API endpoint.

bentoml serve bentoml_service:Summarization

If you’re not comfortable with Swagger UI, we can also use Python requests to hit the API endpoint.

import bentoml

with bentoml.SyncHTTPClient("http://localhost:3000") as client:
    result = client.summarize(
        text="Breaking News: In an astonishing turn of events, the small town of Willow Creek has been taken by storm as local resident Jerry Thompson's cat, Whiskers, performed what witnesses are calling a 'miraculous and gravity-defying leap.' Eyewitnesses report that Whiskers, an otherwise unremarkable tabby cat, jumped a record-breaking 20 feet into the air to catch a fly. The event, which took place in Thompson's backyard, is now being investigated by scientists for potential breaches in the laws of physics. Local authorities are considering a town festival to celebrate what is being hailed as 'The Leap of the Century.'"
    )
    print(result)

Adaptive Batching In BentoML

A major challenge with ML inference is how to utilize the resources effectively, especially the GPU. Handling one request at a time results in reduced usage of GPU RAM, and there is higher latency when multiple requests are sent, and poor throughput. To overcome this, Batching comes into the picture.

Many models achieve higher throughput, better resource utilization, and lower latency when processing requests in batches.

• BentoML supports adaptive batching, a dynamic request dispatching mechanism that intelligently groups multiple requests for more efficient processing.
• It continuously adjusts batch size and window based on real-time traffic patterns.
• This ensures optimal performance as it provides fast responses during low-traffic periods and maximizes resource utilization under heavy load.

Terms around Batching

Batching means grouping multiple inputs into a single batch for processing. It includes two main concepts:

Batch window: Maximum time a service waits to accumulate requests into a batch before processing.

Batch size: Maximum number of requests in a batch.

For multiple Services, the Service responsible for running model inference (ServiceTwo in the diagram below) collects requests from the intermediary Service (ServiceOne) and forms batches based on optimal latency.

The adaptive batching algorithm continuously learns and adjusts the batching parameters based on recent trends in request patterns and processing time. This means that during high traffic time, batches are likely to be larger and processed more frequently, whereas during quieter periods, BentoML will prioritize reducing latency, even if that means smaller batch sizes. The order of the requests in a batch is not guaranteed.

Example Two

Create a BentoML service with adaptive batching enabled.

• We create a runtime environment by using bentoml.images.PythonImage and adding parameters as required.
• While running the bentoml.service , we ensure that we run the service in the runtime defined in previous step.
• By default, adaptive batching is disabled. Use the @bentoml.api decorator to enable it and configure the batch behavior for an API endpoint.
• Note that the batchable API:
• Should be of a type that can encapsulate multiple individual requests, such as typing.List[str] or numpy.ndarray.
• It only accepts one parameter in addition to bentoml.Context.

#bentoml_batching.py

import bentoml
from PIL.Image import Image
import numpy as np
from typing import Dict
from typing import List
from pydantic import Field

MODEL_ID = "openai/clip-vit-base-patch32"

runtime_image = bentoml.images.PythonImage(
    python_version="3.11"
).requirements_file("requirements.txt")

def _resize_img(img: Image):
    return img.resize((224, 224))

@bentoml.service(
    image=runtime_image,
    resources={
        "memory" : "4Gi"
    }
)
class CLIP:

    hf_model = bentoml.models.HuggingFaceModel(MODEL_ID)

    def __init__(self) -> None:
        import torch
        from transformers import CLIPModel, CLIPProcessor
        self.device = "cuda" if torch.cuda.is_available() else "cpu"
        self.model = CLIPModel.from_pretrained(self.hf_model).to(self.device)
        self.processor = CLIPProcessor.from_pretrained(self.hf_model)
        self.logit_scale = self.model.logit_scale.item() if self.model.logit_scale.item() else 4.60517
        print("Model clip loaded", "device:", self.device)

    @bentoml.api(batchable=True)
    async def encode_image(self, items: List[Image]) -> np.ndarray:
        '''
        generate the 512-d embeddings of the images
        '''
        return await self._encode_image(items)

    async def _encode_image(self, items: List[Image]) -> np.ndarray:
        items = [_resize_img(item) for item in items]
        inputs = self.processor(images=items, return_tensors="pt", padding=True).to(self.device)
        image_embeddings = self.model.get_image_features(**inputs)
        return image_embeddings.cpu().detach().numpy()

    @bentoml.api(batchable=True)
    async def encode_text(self, items: List[str]) -> np.ndarray:
        '''
        generate the 512-d embeddings of the texts
        '''
        return await self._encode_text(items)

    async def _encode_text(self, items: List[str]) -> np.ndarray:
        inputs = self.processor(text=items, return_tensors="pt", padding=True).to(self.device)
        text_embeddings = self.model.get_text_features(**inputs)
        return text_embeddings.cpu().detach().numpy()

    @bentoml.api
    async def rank(self, queries: List[Image], candidates : List[str] = Field(["picture of a dog", "picture of a cat"], description="list of description candidates")) -> Dict[str, List[List[float]]]:
        '''
        return the similarity between the query images and the candidate texts
        '''

        # Encode embeddings
        query_embeds = await self._encode_image(queries)
        candidate_embeds = await self._encode_text(candidates)

        # Compute cosine similarities
        cosine_similarities = self.cosine_similarity(query_embeds, candidate_embeds)
        logit_scale = np.exp(self.logit_scale)
        # Compute softmax scores
        prob_scores = self.softmax(logit_scale * cosine_similarities)
        return {
            "probabilities": prob_scores.tolist(),
            "cosine_similarities" : cosine_similarities.tolist(),
        }

    @staticmethod
    def cosine_similarity(query_embeds, candidates_embeds):
        # Normalize each embedding to a unit vector
        query_embeds /= np.linalg.norm(query_embeds, axis=1, keepdims=True)
        candidates_embeds /= np.linalg.norm(candidates_embeds, axis=1, keepdims=True)

        # Compute cosine similarity
        cosine_similarities = np.matmul(query_embeds, candidates_embeds.T)

        return cosine_similarities

    @staticmethod
    def softmax(scores):
        # Compute softmax scores (probabilities)
        exp_scores = np.exp(
            scores - np.max(scores, axis=-1, keepdims=True)
        )  # Subtract max for numerical stability
        return exp_scores / np.sum(exp_scores, axis=-1, keepdims=True)

Run example two:

bentoml serve bentoml_batching:CLIP

Test the model using Python Client:

import requests
import json

url = "http://localhost:3000/rank"

# Open your images in binary mode
files = [
    ("queries", open("dog.png", "rb")),
    ("queries", open("cat.png", "rb")),
]

# The 'candidates' list needs to be sent as a string if using form-data
data = {
    "candidates": json.dumps(["a happy dog", "a grumpy cat"])
}

response = requests.post(url, files=files, data=data)

print(json.dumps(response.json(), indent=2))

Model Composition In BentoML

In BentoML, we can combine multiple models to build complex applications like RAG.

BentoML provides Service APIs for creating workflows, where models need to work

• Either in sequence (one after another)
• or in parallel(at the same time).

When to use model composition:

• Processing different types of data (e.g., text and images)
• Improve accuracy by combining predictions from multiple models
• Run different models on specialized hardware (e.g., GPU for one model, CPU for another)
• Orchestrate sequential steps like preprocessing, inference, and postprocessing with specialized models or services.

Example Three

Run multiple models in one service: You can run multiple models on the same hardware device and expose separate or combined APIs for them.

#bentoml_composition.py
import bentoml
from bentoml.models import HuggingFaceModel
from transformers import pipeline
from typing import List

# Run two models in the same Service on the same hardware device
@bentoml.service(
    resources={"gpu": 1, "memory": "4GiB"},
    traffic={"timeout": 20},
)
class MultiModelService:
    # Retrieve model references from HF by specifying its HF ID
    model_a_path = HuggingFaceModel("FacebookAI/roberta-large-mnli")
    model_b_path = HuggingFaceModel("distilbert/distilbert-base-uncased")

    def __init__(self) -> None:
        # Initialize pipelines for each model
        self.pipeline_a = pipeline(task="zero-shot-classification", model=self.model_a_path, hypothesis_template="This text is about {}")
        self.pipeline_b = pipeline(task="sentiment-analysis", model=self.model_b_path)

    # Define an API for data processing with model A
    @bentoml.api
    def process_a(self, input_data: str, labels: List[str] = ["positive", "negative", "neutral"]) -> dict:
        return self.pipeline_a(input_data, labels)

    # Define an API for data processing with model B
    @bentoml.api
    def process_b(self, input_data: str) -> dict:
        return self.pipeline_b(input_data)[0]

    # Define an API endpoint that combines the processing of both models
    @bentoml.api
    def combined_process(self, input_data: str, labels: List[str] = ["positive", "negative", "neutral"]) -> dict:
        classification = self.pipeline_a(input_data, labels)
        sentiment = self.pipeline_b(input_data)[0]
        return {
            "classification": classification,
            "sentiment": sentiment
        }

A few things to note

• Making use of GPU resources by assigning them while creating a service.
• Creating multiple independent APIs inside a single bentoml service.
• Combining multiple API as a single API using combined_process

Digital Products

Losing time scrolling through socials, use *Social Media Time Tracker: Take Back Your Time*

Connect with the author here:

LinkedIn | YouTube | Threads | Twitter | Instagram | Facebook

References

[embed]BentoML Documentation github_stars pypi_status actions_status documentation_status join_slack BentoML is a Unified Inference Platform for…docs.bentoml.com