Tutorial LitServe: Framework Serving Model AI yang Cepat dan Mudah

Pendahuluan

LitServe adalah framework open-source dari Lightning AI yang dirancang untuk menyajikan (serving) model AI/ML dengan performa tinggi dan kemudahan penggunaan. Dibangun di atas FastAPI, LitServe menyediakan fitur-fitur enterprise seperti batching, streaming, GPU autoscaling, dan multi-model endpoints tanpa memerlukan konfigurasi yang rumit.

Berbeda dengan framework serving lainnya yang memerlukan banyak boilerplate code dan konfigurasi infrastruktur, LitServe memungkinkan Anda mengubah model AI menjadi API production-ready hanya dengan beberapa baris kode. Framework ini mendukung semua jenis model termasuk LLM, computer vision, NLP, audio, dan model klasik scikit-learn.

Dalam tutorial ini, kita akan mempelajari cara menggunakan LitServe mulai dari instalasi, membuat API sederhana, hingga fitur-fitur lanjutan seperti batching, streaming, dan authentication.

Mengapa LitServe?

Sebelum masuk ke implementasi, berikut alasan mengapa LitServe layak dipertimbangkan:

• Performa Tinggi: Hingga 2x lebih cepat dibanding FastAPI biasa berkat optimasi internal
• Minimal Boilerplate: Cukup definisikan setup(), decoderequest(), predict(), dan encoderesponse()
• Batching Otomatis: Menggabungkan multiple request untuk throughput lebih tinggi pada GPU
• Streaming Response: Dukungan native untuk streaming output (cocok untuk LLM)
• Multi-Model: Satu server bisa melayani beberapa model sekaligus
• GPU Management: Otomatis menangani alokasi GPU dan multi-worker
• OpenAPI Docs: Otomatis menghasilkan dokumentasi API (Swagger UI)

Instalasi

Prasyarat

Pastikan Anda memiliki Python 3.8 atau lebih baru. Disarankan menggunakan virtual environment.

python -m venv litserve-env
source litserve-env/bin/activate  # Linux/Mac
atau
litserve-env\Scripts\activate     # Windows

Install LitServe

pip install litserve

Untuk fitur tambahan, Anda bisa menginstall dependensi opsional:

# Untuk dukungan semua fitur
pip install litserve[all]

Untuk model PyTorch
pip install torch torchvision

Untuk model Hugging Face
pip install transformers

Untuk model scikit-learn
pip install scikit-learn

Verifikasi instalasi:

import litserve as ls
print(ls.version)

Basic Usage: Membuat API Pertama

Konsep Dasar: LitAPI

Inti dari LitServe adalah class LitAPI. Anda perlu mengimplementasikan minimal 4 method:

Contoh 1: API Sederhana dengan Scikit-Learn

Mari buat API untuk model klasifikasi iris menggunakan scikit-learn.

# server.py
import litserve as ls
from sklearn.datasets import loadiris

from sklearn.ensemble import RandomForestClassifier
import numpy as np

class IrisAPI(ls.LitAPI):
    def setup(self, device):
        """Load dan train model saat server startup."""
        iris = loadiris()
        self.model = RandomForestClassifier(nestimators=100, randomstate=42)
        self.model.fit(iris.data, iris.target)
        self.classnames = iris.targetnames.tolist()

    def decoderequest(self, request):

        """Parse input dari HTTP request."""
        return np.array(request["features"]).reshape(1, -1)

    def predict(self, x):
        """Jalankan prediksi."""
        prediction = self.model.predict(x)
        probability = self.model.predictproba(x)
        return {
            "classidx": int(prediction[0]),

            "probabilities": probability[0].tolist()
        }

    def encoderesponse(self, output):
        """Format response untuk client."""
        return {
            "predictedclass": self.classnames[output["classidx"]],

            "probabilities": {
                name: round(prob, 4)
                for name, prob in zip(self.classnames, output["probabilities"])
            }
        }

if name == "main":
    api = IrisAPI()
    server = ls.LitServer(api)
    server.run(port=8000)

Jalankan server:

python server.py

Test dengan curl:

curl -X POST http://localhost:8000/predict \
  -H "Content-Type: application/json" \
  -d '{"features": [5.1, 3.5, 1.4, 0.2]}'

Response:

{
  "predictedclass": "setosa",

  "probabilities": {
    "setosa": 0.9800,
    "versicolor": 0.0150,
    "virginica": 0.0050
  }
}

# visionserver.py
import litserve as ls
import torch
from torchvision import transforms, models
from PIL import Image
import io
import base64

class ImageClassifierAPI(ls.LitAPI):
    def setup(self, device):
        """Load pretrained ResNet model."""
        self.device = device
        self.model = models.resnet50(weights=models.ResNet50Weights.DEFAULT)

        self.model.to(device)
        self.model.eval()

        self.transform = transforms.Compose([
            transforms.Resize(256),
            transforms.CenterCrop(224),
            transforms.ToTensor(),
            transforms.Normalize(
                mean=[0.485, 0.456, 0.406],
                std=[0.229, 0.224, 0.225]
            ),
        ])

        # Load ImageNet labels
        self.labels = models.ResNet50Weights.DEFAULT.meta["categories"]

    def decoderequest(self, request):

        """Decode base64 image dari request."""
        imagebytes = base64.b64decode(request["image"])
        image = Image.open(io.BytesIO(imagebytes)).convert("RGB")

        return self.transform(image).unsqueeze(0).to(self.device)

    def predict(self, x):
        """Jalankan inferensi."""
        with torch.nograd():
            outputs = self.model(x)
            probabilities = torch.nn.functional.softmax(outputs[0], dim=0)
        return probabilities

    def encoderesponse(self, output):

        """Return top-5 predictions."""
        top5prob, top5idx = torch.topk(output, 5)

        results = []
        for i in range(5):
            results.append({
                "label": self.labels[top5idx[i].item()],
                "confidence": round(top5prob[i].item(), 4)

            })
        return {"predictions": results}

if name == "main":
    api = ImageClassifierAPI()
    server = ls.LitServer(api, accelerator="auto")
    server.run(port=8000)

import litserve as ls
import torch
from transformers import AutoTokenizer, AutoModel

class EmbeddingAPI(ls.LitAPI):
    def setup(self, device):
        self.device = device
        modelname = "sentence-transformers/all-MiniLM-L6-v2"
        self.tokenizer = AutoTokenizer.frompretrained(modelname)
        self.model = AutoModel.frompretrained(modelname).to(device)
        self.model.eval()

    def decoderequest(self, request):

        return request["text"]

    def predict(self, batch):
        """Dengan batching, batch adalah list dari semua input."""
        encoded = self.tokenizer(
            batch,
            padding=True,
            truncation=True,
            maxlength=512,
            returntensors="pt"

        ).to(self.device)

        with torch.nograd():
            outputs = self.model(*encoded)


        # Mean pooling
        attentionmask = encoded["attentionmask"]
        tokenembeddings = outputs.lasthiddenstate

        inputmask = attentionmask.unsqueeze(-1).expand(tokenembeddings.size()).float()
        embeddings = torch.sum(tokenembeddings  inputmask, 1) / torch.clamp(
            inputmask.sum(1), min=1e-9

        )

        return embeddings.cpu().tolist()

    def encoderesponse(self, output):
        return {"embedding": output}

if name == "main":
    api = EmbeddingAPI()
    server = ls.LitServer(
        api,
        accelerator="auto",
        maxbatchsize=32,      # Maksimal 32 request per batch
        batchtimeout=0.05,     # Tunggu 50ms untuk mengumpulkan batch

    )
    server.run(port=8000)

import litserve as ls
from transformers import AutoTokenizer, AutoModelForCausalLM
import torch

class LLMStreamingAPI(ls.LitAPI):
    def setup(self, device):
        self.device = device
        modelname = "TinyLlama/TinyLlama-1.1B-Chat-v1.0"
        self.tokenizer = AutoTokenizer.frompretrained(modelname)
        self.model = AutoModelForCausalLM.frompretrained(

            modelname,
            torchdtype=torch.float16

        ).to(device)

    def decoderequest(self, request):
        prompt = request["prompt"]
        maxtokens = request.get("maxtokens", 256)
        return {"prompt": prompt, "maxtokens": maxtokens}

    def predict(self, x):
        """Generator function untuk streaming."""
        inputs = self.tokenizer(
            x["prompt"],
            returntensors="pt"

        ).to(self.device)

        generatedids = inputs["inputids"]


        for  in range(x["maxtokens"]):

            with torch.nograd():
                outputs = self.model(generatedids)


            nexttokenlogits = outputs.logits[:, -1, :]

            nexttokenid = torch.argmax(nexttokenlogits, dim=-1, keepdim=True)


            if nexttokenid.item() == self.tokenizer.eostokenid:

                break

            generatedids = torch.cat([generatedids, nexttokenid], dim=-1)

            tokentext = self.tokenizer.decode(
                nexttokenid[0],
                skipspecialtokens=True
            )

            yield tokentext


    def encoderesponse(self, output):
        for token in output:
            yield {"token": token}

if name == "main":
    api = LLMStreamingAPI()
    server = ls.LitServer(api, stream=True, accelerator="auto")
    server.run(port=8000)

Client untuk streaming:

import requests

response = requests.post(
    "http://localhost:8000/predict",
    json={"prompt": "Explain quantum computing in simple terms:", "maxtokens": 100},

    stream=True
)

for chunk in response.itercontent(chunksize=None):

    if chunk:
        print(chunk.decode(), end="", flush=True)

import litserve as ls
import torch
from transformers import pipeline

class SentimentAPI(ls.LitAPI):
    def setup(self, device):
        self.pipe = pipeline(
            "sentiment-analysis",
            model="distilbert-base-uncased-finetuned-sst-2-english",
            device=device
        )

    def decoderequest(self, request):
        return request["text"]

    def predict(self, x):
        return self.pipe(x)[0]

    def encoderesponse(self, output):

        return {"label": output["label"], "score": round(output["score"], 4)}


class SummarizationAPI(ls.LitAPI):
    def setup(self, device):
        self.pipe = pipeline(
            "summarization",
            model="sshleifer/distilbart-cnn-12-6",
            device=device
        )

    def decoderequest(self, request):
        return {
            "text": request["text"],
            "maxlength": request.get("maxlength", 130)
        }

    def predict(self, x):
        return self.pipe(
            x["text"],
            maxlength=x["maxlength"],
            minlength=30

        )[0]

    def encoderesponse(self, output):
        return {"summary": output["summarytext"]}



if name == "main":
    sentimentserver = ls.LitServer(SentimentAPI(), accelerator="auto")
    summaryserver = ls.LitServer(SummarizationAPI(), accelerator="auto")


    # Jalankan keduanya
    server = ls.LitServer(
        [SentimentAPI(), SummarizationAPI()],
        apipath=["/sentiment", "/summarize"],
        accelerator="auto"
    )
    server.run(port=8000)

Authentication dan Middleware

Lindungi API Anda dengan authentication.

import litserve as ls

class ProtectedAPI(ls.LitAPI):
    def setup(self, device):
        self.model = lambda x: x * 2  # Model sederhana

    def decoderequest(self, request):

        return request["value"]

    def predict(self, x):
        return self.model(x)

    def encoderesponse(self, output):
        return {"result": output}

    def authorize(self, authheader):

        """Validasi API key dari header Authorization."""
        if authheader != "Bearer my-secret-key":
            raise ls.HTTPException(statuscode=401, detail="Invalid API key")


if name == "main":
    api = ProtectedAPI()
    server = ls.LitServer(api)
    server.run(port=8000)

if name == "main":
    api = MyModelAPI()
    server = ls.LitServer(
        api,
        accelerator="gpu",
        devices="auto",        # Gunakan semua GPU yang tersedia
        workersperdevice=2,  # 2 worker per GPU

        timeout=30,            # Timeout 30 detik per request
    )
    server.run(port=8000)

server = ls.LitServer(
    api,
    accelerator="gpu",
    devices=[0, 1, 2, 3],    # Gunakan GPU 0-3
    workersperdevice=1,

)

from pydantic import BaseModel
from typing import List

class PredictRequest(BaseModel):
    features: List[float]
    modelversion: str = "v1"

class MyAPI(ls.LitAPI):
    def decoderequest(self, request: PredictRequest):

        return request.features

curl http://localhost:8000/health
Response: {"status": "ok"}

class RobustAPI(ls.LitAPI):
    def decoderequest(self, request):
        if "text" not in request:
            raise ls.HTTPException(
                statuscode=400,

                detail="Field 'text' is required"
            )
        if len(request["text"]) > 10000:
            raise ls.HTTPException(
                statuscode=400,
                detail="Text exceeds maximum length of 10000 characters"
            )
        return request["text"]

4. Logging dan Debugging

import logging

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

class DebugAPI(ls.LitAPI):
    def predict(self, x):
        logger.info(f"Processing input of shape: {x.shape}")
        result = self.model(x)
        logger.info(f"Prediction complete: {result[:5]}...")
        return result

5. Optimasi Memory

class MemoryEfficientAPI(ls.LitAPI):
    def setup(self, device):
        self.model = loadmodel()

        # Gunakan half precision untuk menghemat VRAM
        if "cuda" in str(device):
            self.model = self.model.half()
        self.model.eval()

    def predict(self, x):
        with torch.nograd(), torch.cuda.amp.autocast():
            return self.model(x)

6. Docker Deployment

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY server.py .

EXPOSE 8000
CMD ["python", "server.py"]

# requirements.txt
litserve
torch
transformers

Testing API

Unit Testing

import pytest
from server import IrisAPI

def testirisapi():
    api = IrisAPI()
    api.setup("cpu")

    # Test decode
    request = {"features": [5.1, 3.5, 1.4, 0.2]}
    x = api.decoderequest(request)

    assert x.shape == (1, 4)

    # Test predict
    result = api.predict(x)
    assert "classidx" in result
    assert "probabilities" in result

    # Test encode
    response = api.encoderesponse(result)

    assert "predictedclass" in response
    assert response["predictedclass"] in ["setosa", "versicolor", "virginica"]

# locustfile.py
from locust import HttpUser, task, between

class APIUser(HttpUser):
    waittime = between(0.1, 0.5)

    @task
    def predict(self):
        self.client.post("/predict", json={
            "features": [5.1, 3.5, 1.4, 0.2]
        })

pip install locust
locust -f locustfile.py --host=http://localhost:8000

Perbandingan dengan Framework Lain

| Fitur | LitServe | FastAPI | BentoML | TorchServe |

|-------|----------|---------|---------|------------|

| Setup Complexity | Rendah | Rendah | Sedang | Tinggi |

| Auto Batching | Ya | Tidak | Ya | Ya |

| Streaming | Ya | Ya | Ya | Ya |

| Multi-GPU | Ya | Manual | Ya | Ya |

| OpenAPI Docs | Ya | Ya | Ya | Tidak |

| GPU Autoscaling | Ya | Tidak | Ya | Terbatas |

Kesimpulan

LitServe menyederhanakan proses deployment model AI dengan menyediakan abstraksi yang tepat tanpa mengorbankan fleksibilitas. Dengan hanya mengimplementasikan 4 method dasar (setup, decoderequest, predict, encoderesponse), Anda bisa membuat API production-ready yang mendukung batching, streaming, multi-model serving, dan GPU scaling.

Poin-poin penting yang perlu diingat:

• Mulai dengan implementasi LitAPI sederhana, lalu tambahkan fitur sesuai kebutuhan
• Gunakan batching untuk meningkatkan throughput pada GPU
• Manfaatkan streaming untuk model generatif
• Terapkan authentication untuk API production
• Lakukan load testing sebelum deployment ke production

LitServe cocok untuk tim yang ingin fokus pada pengembangan model tanpa harus menghabiskan waktu pada infrastruktur serving yang kompleks. Dengan dukungan dari Lightning AI, framework ini terus berkembang dan mendapat update regular.