Tutorial Lengkap BentoML: Packaging dan Serving ML Models ke Production

BentoML adalah framework open-source untuk building, shipping, dan scaling AI applications. Dengan BentoML, Anda dapat mengubah model ML menjadi production-ready API services dengan mudah, lengkap dengan containerization, batching, dan monitoring.

Mengapa BentoML?

Tantangan dalam ML deployment:

• Packaging complexity: Bundling model dengan dependencies
• Serving infrastructure: Setup web server, API endpoints
• Performance: Batching, caching, GPU utilization
• Scalability: Horizontal scaling, load balancing
• Multi-framework: Support berbagai ML frameworks

• Unified API untuk semua ML frameworks
• Auto-generated REST/gRPC APIs
• Built-in adaptive batching
• Docker/Kubernetes deployment ready
• Model versioning dan management

Instalasi

# Install BentoML
pip install bentoml

Dengan framework-specific support
pip install "bentoml[pytorch]"
pip install "bentoml[tensorflow]"
pip install "bentoml[sklearn]"
pip install "bentoml[transformers]"

Verify installation
bentoml --version

Quick Start

1. Save Model ke BentoML

# trainandsave.py
import bentoml
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import loadiris


Train model
X, y = loadiris(returnXy=True)
model = RandomForestClassifier(nestimators=100)

model.fit(X, y)

Save model ke BentoML
savedmodel = bentoml.sklearn.savemodel(

    "irisclassifier",
    model,
    signatures={
        "predict": {"batchable": True, "batchdim": 0}

    },
    labels={"framework": "sklearn", "dataset": "iris"},
    metadata={"accuracy": 0.97}
)

print(f"Model saved: {savedmodel}")
Output: Model(tag="irisclassifier:abc123")

# service.py
import numpy as np
import bentoml
from bentoml.io import NumpyNdarray, JSON

Load model
irismodel = bentoml.sklearn.get("irisclassifier:latest")


Create runner
irisrunner = irismodel.torunner()

Create service
svc = bentoml.Service("irisservice", runners=[irisrunner])

Define API endpoint
@svc.api(input=NumpyNdarray(), output=NumpyNdarray())
async def predict(inputarray: np.ndarray) -> np.ndarray:

    return await irisrunner.predict.asyncrun(inputarray)

Alternative: JSON input/output
@svc.api(input=JSON(), output=JSON())
async def classify(inputdata: dict) -> dict:

    features = np.array(inputdata["features"]).reshape(1, -1)
    prediction = await irisrunner.predict.asyncrun(features)
    classnames = ["setosa", "versicolor", "virginica"]

    return {
        "prediction": int(prediction[0]),
        "classname": classnames[prediction[0]]

    }

# Development server
bentoml serve service:svc --reload

Production server
bentoml serve service:svc --production

Specify port
bentoml serve service:svc --port 3000

# testservice.py
import requests
import numpy as np

Test dengan NumpyNdarray
data = np.array([[5.1, 3.5, 1.4, 0.2]])
response = requests.post(
    "http://localhost:3000/predict",
    headers={"content-type": "application/json"},
    json=data.tolist()
)
print(f"Prediction: {response.json()}")

Test dengan JSON
response = requests.post(
    "http://localhost:3000/classify",
    json={"features": [5.1, 3.5, 1.4, 0.2]}
)
print(f"Classification: {response.json()}")

Building Bentos

1. Buat bentofile.yaml

# bentofile.yaml
service: "service:svc"
labels:
  owner: ml-team
  project: iris-classifier
include:
".py"

python:

  packages:
scikit-learn
numpy
docker:

  distro: debian
  pythonversion: "3.10"

# Build bento
bentoml build

Output:
Building BentoML service "irisservice:abc123"
Bento(tag="irisservice:abc123") built successfully


List bentos
bentoml list

Get bento info
bentoml get irisservice:latest

# Build Docker image
bentoml containerize irisservice:latest


Atau dengan custom tag
bentoml containerize irisservice:latest -t my-registry/iris-service:v1

Run container
docker run -p 3000:3000 irisservice:abc123

# multimodelservice.py
import bentoml
import numpy as np
from bentoml.io import NumpyNdarray, JSON

Load multiple models
classifier = bentoml.sklearn.get("classifier:latest")
regressor = bentoml.sklearn.get("regressor:latest")

Create runners
clfrunner = classifier.torunner()

regrunner = regressor.torunner()


Create service with multiple runners
svc = bentoml.Service("multimodelservice", runners=[clfrunner, regrunner])


@svc.api(input=NumpyNdarray(), output=JSON())
async def predictall(inputarray: np.ndarray) -> dict:

    # Run both models in parallel
    classification = await clfrunner.predict.asyncrun(inputarray)
    regression = await regrunner.predict.asyncrun(inputarray)


    return {
        "classification": classification.tolist(),
        "regression": regression.tolist()
    }

import bentoml
import numpy as np

class MyCustomRunner(bentoml.Runnable):
    SUPPORTEDRESOURCES = ("cpu",)
    SUPPORTSCPUMULTITHREADING = True


    def init(self):
        # Initialize your custom logic
        self.model = self.loadmodel()


    def loadmodel(self):

        # Load your model
        pass

    @bentoml.Runnable.method(batchable=True, batchdim=0)
    def predict(self, inputarray: np.ndarray) -> np.ndarray:

        return self.model.predict(inputarray)

    @bentoml.Runnable.method(batchable=False)
    def preprocess(self, data: dict) -> np.ndarray:
        # Custom preprocessing
        return np.array(data["features"])

Create runner
myrunner = bentoml.Runner(MyCustomRunner, name="myrunner")
svc = bentoml.Service("customservice", runners=[myrunner])

import bentoml
from bentoml.io import NumpyNdarray
import numpy as np

model = bentoml.sklearn.get("irisclassifier:latest")


Configure batching
runner = model.torunner()
runner.configure(
    maxbatchsize=32,
    maxlatencyms=100,
)

svc = bentoml.Service("batchedservice", runners=[runner])


@svc.api(
    input=NumpyNdarray(),
    output=NumpyNdarray(),
)
async def predict(inputarray: np.ndarray) -> np.ndarray:
    # BentoML automatically batches concurrent requests
    return await runner.predict.asyncrun(inputarray)

import bentoml
from bentoml.io import (
    NumpyNdarray,
    PandasDataFrame,
    JSON,
    Text,
    Image,
    File,
    Multipart
)
import numpy as np
import pandas as pd
from PIL import Image as PILImage

svc = bentoml.Service("ioexamples")


NumpyNdarray
@svc.api(input=NumpyNdarray(dtype="float32", shape=(-1, 4)), output=NumpyNdarray())
async def numpyendpoint(arr: np.ndarray) -> np.ndarray:
    return arr  2

PandasDataFrame
@svc.api(input=PandasDataFrame(), output=PandasDataFrame())
async def pandasendpoint(df: pd.DataFrame) -> pd.DataFrame:

    df["processed"] = True
    return df

JSON
@svc.api(input=JSON(), output=JSON())
async def jsonendpoint(data: dict) -> dict:
    return {"result": data.get("value", 0) * 2}

Text
@svc.api(input=Text(), output=Text())
async def textendpoint(text: str) -> str:

    return text.upper()

Image
@svc.api(input=Image(), output=JSON())
async def imageendpoint(img: PILImage.Image) -> dict:
    return {"width": img.width, "height": img.height}

File
@svc.api(input=File(), output=JSON())
async def fileendpoint(file: bytes) -> dict:

    return {"size": len(file)}

# pytorchservice.py
import torch
import torch.nn as nn
import bentoml
from bentoml.io import NumpyNdarray
import numpy as np

Define model
class SimpleNet(nn.Module):
    def init(self):
        super().init()
        self.fc = nn.Linear(4, 3)

    def forward(self, x):
        return self.fc(x)

Save model
model = SimpleNet()
savedmodel = bentoml.pytorch.savemodel(
    "pytorchmodel",

    model,
    signatures={"call": {"batchable": True}}
)

Service
pytorchrunner = bentoml.pytorch.get("pytorchmodel:latest").torunner()
svc = bentoml.Service("pytorchservice", runners=[pytorchrunner])

@svc.api(input=NumpyNdarray(dtype="float32"), output=NumpyNdarray())
async def predict(inputarray: np.ndarray) -> np.ndarray:

    tensor = torch.fromnumpy(inputarray)

    output = await pytorchrunner.asyncrun(tensor)

    return output.numpy()

# tensorflowservice.py
import tensorflow as tf
import bentoml
from bentoml.io import NumpyNdarray
import numpy as np

Create and save model
model = tf.keras.Sequential([
    tf.keras.layers.Dense(64, activation='relu', inputshape=(4,)),

    tf.keras.layers.Dense(3, activation='softmax')
])
model.compile(optimizer='adam', loss='sparsecategoricalcrossentropy')


Save to BentoML
savedmodel = bentoml.keras.savemodel("kerasmodel", model)

Service
kerasrunner = bentoml.keras.get("kerasmodel:latest").torunner()

svc = bentoml.Service("kerasservice", runners=[kerasrunner])


@svc.api(input=NumpyNdarray(), output=NumpyNdarray())
async def predict(inputarray: np.ndarray) -> np.ndarray:
    return await kerasrunner.predict.asyncrun(inputarray)

# transformersservice.py
import bentoml
from bentoml.io import Text, JSON
from transformers import pipeline

Save pipeline
classifier = pipeline("sentiment-analysis")
savedmodel = bentoml.transformers.savemodel(
    "sentimentclassifier",

    classifier
)

Service
sentimentrunner = bentoml.transformers.get("sentimentclassifier:latest").torunner()
svc = bentoml.Service("sentimentservice", runners=[sentimentrunner])

@svc.api(input=Text(), output=JSON())
async def analyze(text: str) -> dict:
    result = await sentimentrunner.asyncrun(text)
    return result[0]

4. ONNX

# onnxservice.py
import bentoml
from bentoml.io import NumpyNdarray
import numpy as np

Save ONNX model
savedmodel = bentoml.onnx.savemodel(

    "onnxmodel",
    "model.onnx"
)

Service
onnxrunner = bentoml.onnx.get("onnxmodel:latest").torunner()

svc = bentoml.Service("onnxservice", runners=[onnxrunner])


@svc.api(input=NumpyNdarray(dtype="float32"), output=NumpyNdarray())
async def predict(inputarray: np.ndarray) -> np.ndarray:
    return await onnxrunner.run.asyncrun(inputarray)

# bentofile.yaml
service: "service:svc"
python:
  packages:
scikit-learn
numpy
docker:

  distro: debian
  pythonversion: "3.10"
  cudaversion: "11.8"  # For GPU

  env:
BENTOMLCONFIG=/app/config.yaml
  setupscript: "./setup.sh"

# Build dan push
bentoml containerize irisservice:latest -t myregistry/iris:v1
docker push myregistry/iris:v1

Run dengan GPU
docker run --gpus all -p 3000:3000 myregistry/iris:v1

2. Kubernetes Deployment

# k8s-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: iris-service
spec:
  replicas: 3
  selector:
    matchLabels:
      app: iris-service
  template:
    metadata:
      labels:
        app: iris-service
    spec:
      containers:
name: iris-service
          image: myregistry/iris:v1

          ports:
containerPort: 3000
          resources:

            requests:
              memory: "512Mi"
              cpu: "500m"
            limits:
              memory: "1Gi"
              cpu: "1000m"
          readinessProbe:
            httpGet:
              path: /healthz
              port: 3000
            initialDelaySeconds: 5
            periodSeconds: 10

apiVersion: v1
kind: Service
metadata:
  name: iris-service
spec:
  selector:
    app: iris-service
  ports:
port: 80
      targetPort: 3000

  type: LoadBalancer

3. BentoCloud (Managed)

# Login ke BentoCloud
bentoml cloud login

Deploy
bentoml deploy irisservice:latest


List deployments
bentoml deployment list

Get deployment info
bentoml deployment get iris-service

# servicewithmetrics.py
import bentoml
from bentoml.io import JSON
import prometheusclient

Custom metrics
REQUESTCOUNT = prometheusclient.Counter(
    "predictionrequeststotal",
    "Total prediction requests"
)
PREDICTIONLATENCY = prometheusclient.Histogram(
    "predictionlatencyseconds",
    "Prediction latency"
)

svc = bentoml.Service("monitoredservice")


@svc.api(input=JSON(), output=JSON())
async def predict(data: dict) -> dict:
    REQUESTCOUNT.inc()
    with PREDICTIONLATENCY.time():

        # Your prediction logic
        result = {"prediction": 1}
    return result

import logging
import bentoml

Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(name)

svc = bentoml.Service("loggedservice")

@svc.api(input=JSON(), output=JSON())
async def predict(data: dict) -> dict:
    logger.info(f"Received request: {data}")

    try:
        result = process(data)
        logger.info(f"Prediction successful: {result}")
        return result
    except Exception as e:
        logger.error(f"Prediction failed: {e}")
        raise

3. Tracing dengan OpenTelemetry

# bentofile.yaml
service: "service:svc"
python:
  packages:
opentelemetry-api
opentelemetry-sdk
opentelemetry-exporter-jaeger

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.jaeger.thrift import JaegerExporter

Setup tracing
trace.settracerprovider(TracerProvider())
jaegerexporter = JaegerExporter(

    agenthostname="localhost",

    agentport=6831,
)
trace.gettracerprovider().addspanprocessor(
    BatchSpanProcessor(jaegerexporter)

)

tracer = trace.gettracer(name)

@svc.api(input=JSON(), output=JSON())
async def predict(data: dict) -> dict:
    with tracer.startascurrentspan("prediction"):

        result = await runner.predict.asyncrun(data)
    return result

Best Practices

1. Model Versioning

# Save dengan version tag
bentoml.sklearn.savemodel(

    "mymodel",
    model,
    labels={
        "version": "1.0.0",
        "experimentid": "exp-123"

    }
)

Load specific version
model = bentoml.sklearn.get("mymodel:abc123")

List versions
bentoml.models.list("mymodel")

from bentoml.exceptions import BentoMLException

@svc.api(input=JSON(), output=JSON())
async def predict(data: dict) -> dict:
    try:
        if "features" not in data:
            raise ValueError("Missing 'features' field")

        result = await runner.predict.asyncrun(data["features"])
        return {"success": True, "prediction": result}

    except ValueError as e:
        return {"success": False, "error": str(e)}
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        raise BentoMLException(f"Prediction failed: {e}")

3. Testing

# testservice.py
import pytest
import numpy as np
from service import svc

@pytest.fixture
def client():
    return svc.toasyncclient()


@pytest.mark.asyncio
async def testpredict(client):
    inputdata = np.array([[5.1, 3.5, 1.4, 0.2]])

    result = await client.predict(inputdata)
    assert result.shape == (1,)
    assert result[0] in [0, 1, 2]

@pytest.mark.asyncio
async def testclassify(client):

    result = await client.classify({"features": [5.1, 3.5, 1.4, 0.2]})
    assert "prediction" in result
    assert "classname" in result

Kesimpulan

BentoML menyederhanakan proses deployment ML models dengan:

Key takeaways:

• Gunakan BentoML untuk standardize ML deployment
• Leverage adaptive batching untuk throughput
• Containerize untuk consistent deployments
• Implement proper monitoring dan logging
• Test services sebelum production