Complete Ray Serve Tutorial: Scalable ML Model Serving

Ray Serve is a scalable model serving library built on Ray. It enables you to serve ML models with automatic scaling, batching, and multi-model composition, making it ideal for production ML deployments.

Why Ray Serve?

• Framework agnostic: Works with any ML framework
• Scalable: Automatic scaling based on load
• Composable: Combine multiple models easily
• Batching: Automatic request batching
• Native Python: Simple Python-first API

• Model serving at scale
• Multi-model pipelines
• A/B testing
• Real-time inference
• Batch inference

Installation

pip install "ray[serve]"

Verify installation
python -c "import ray; from ray import serve; print(ray.version)"

Quick Start

1. Basic Deployment

from ray import serve
import ray

ray.init()
serve.start()

@serve.deployment
class ModelDeployment:
    def init(self):
        self.model = "simplemodel"


    def call(self, request):
        return {"message": f"Processed by {self.model}"}

Deploy
ModelDeployment.deploy()

Test
import requests
response = requests.get("http://localhost:8000/ModelDeployment")
print(response.json())

from ray import serve
from fastapi import FastAPI
import ray

app = FastAPI()

@serve.deployment
@serve.ingress(app)
class MLService:
    def init(self):
        self.model = self.loadmodel()


    def loadmodel(self):

        return "mymodel"

    @app.get("/predict")
    def predict(self, text: str):
        return {"prediction": f"Result for: {text}"}

    @app.get("/health")
    def health(self):
        return {"status": "healthy"}

ray.init()
serve.run(MLService.bind())

3. Serve ML Model

from ray import serve
import ray
import pickle
import numpy as np

@serve.deployment
class SklearnModel:
    def init(self, modelpath: str):

        with open(modelpath, "rb") as f:
            self.model = pickle.load(f)

    async def call(self, request):
        data = await request.json()
        features = np.array(data["features"]).reshape(1, -1)
        prediction = self.model.predict(features)
        return {"prediction": prediction.tolist()}

ray.init()
serve.run(SklearnModel.bind(modelpath="model.pkl"))

from ray import serve

@serve.deployment(
    numreplicas=3,
    rayactoroptions={
        "numcpus": 2,

        "numgpus": 1,
        "memory": 4  1024  1024  1024  # 4GB

    }
)
class GPUModel:
    def init(self):
        import torch
        self.device = torch.device("cuda")
        self.model = self.loadmodel()

    def loadmodel(self):
        import torch
        model = torch.nn.Linear(10, 2)
        return model.to(self.device)

    async def call(self, request):
        import torch
        data = await request.json()
        tensor = torch.tensor(data["input"]).to(self.device)
        output = self.model(tensor)
        return {"output": output.cpu().tolist()}

from ray import serve
from ray.serve.config import AutoscalingConfig

@serve.deployment(
    autoscalingconfig=AutoscalingConfig(

        minreplicas=1,
        maxreplicas=10,

        targetnumongoingrequestsperreplica=5,
        upscaledelays=10,
        downscaledelays=30
    )
)
class AutoscaledModel:
    def init(self):
        self.model = "autoscaledmodel"


    async def call(self, request):
        import asyncio
        await asyncio.sleep(0.1)  # Simulate processing
        return {"result": "processed"}

from ray import serve
import numpy as np

@serve.deployment
class BatchedModel:
    def init(self):
        self.model = self.loadmodel()


    def loadmodel(self):

        return "batchmodel"

    @serve.batch(maxbatchsize=32, batchwaittimeouts=0.1)

    async def handlebatch(self, inputs: list):
        # Process entire batch at once
        results = [f"processed{i}" for i in range(len(inputs))]

        return results

    async def call(self, request):
        data = await request.json()
        return await self.handlebatch(data["input"])

from ray import serve
import ray

@serve.deployment
class Preprocessor:
    def call(self, text: str) -> str:
        return text.lower().strip()

@serve.deployment
class Model:
    def call(self, text: str) -> dict:
        return {"label": "positive", "score": 0.95}

@serve.deployment
class Postprocessor:
    def call(self, result: dict) -> dict:
        result["processed"] = True
        return result

@serve.deployment
class Pipeline:
    def init(self, preprocessor, model, postprocessor):
        self.preprocessor = preprocessor
        self.model = model
        self.postprocessor = postprocessor

    async def call(self, request):
        data = await request.json()
        text = data["text"]

        # Sequential processing
        cleaned = await self.preprocessor.remote(text)
        prediction = await self.model.remote(cleaned)
        result = await self.postprocessor.remote(prediction)

        return result

Compose
preprocessor = Preprocessor.bind()
model = Model.bind()
postprocessor = Postprocessor.bind()
pipeline = Pipeline.bind(preprocessor, model, postprocessor)

serve.run(pipeline)

from ray import serve
import ray

@serve.deployment
class ModelA:
    def call(self, text: str) -> float:
        return 0.8

@serve.deployment
class ModelB:
    def call(self, text: str) -> float:
        return 0.7

@serve.deployment
class ModelC:
    def call(self, text: str) -> float:
        return 0.9

@serve.deployment
class Ensemble:
    def init(self, modela, modelb, modelc):

        self.models = [modela, modelb, modelc]

    async def call(self, request):
        data = await request.json()
        text = data["text"]

        # Parallel inference
        futures = [model.remote(text) for model in self.models]
        results = await asyncio.gather([ray.get(f) for f in futures])

        # Aggregate results
        avgscore = sum(results) / len(results)

        return {"ensemblescore": avgscore, "individualscores": results}

modela = ModelA.bind()

modelb = ModelB.bind()
modelc = ModelC.bind()

ensemble = Ensemble.bind(modela, modelb, modelc)

serve.run(ensemble)

3. Dynamic Routing

from ray import serve

@serve.deployment
class TextModel:
    def call(self, text: str) -> dict:
        return {"type": "text", "result": "textprediction"}


@serve.deployment
class ImageModel:
    def call(self, image: bytes) -> dict:
        return {"type": "image", "result": "imageprediction"}

@serve.deployment
class Router:
    def init(self, textmodel, imagemodel):
        self.textmodel = textmodel
        self.imagemodel = imagemodel

    async def call(self, request):
        data = await request.json()
        inputtype = data.get("type", "text")


        if inputtype == "text":
            return await self.textmodel.remote(data["content"])

        elif inputtype == "image":
            return await self.imagemodel.remote(data["content"])

        else:
            return {"error": "Unknown input type"}

textmodel = TextModel.bind()
imagemodel = ImageModel.bind()

router = Router.bind(textmodel, imagemodel)


serve.run(router)

from ray import serve
import torch
import torch.nn as nn

@serve.deployment(rayactoroptions={"numgpus": 1})
class PyTorchModel:
    def init(self, modelpath: str):

        self.device = torch.device("cuda" if torch.cuda.isavailable() else "cpu")
        self.model = torch.load(modelpath)

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

    async def call(self, request):
        data = await request.json()
        tensor = torch.tensor(data["input"]).to(self.device)

        with torch.nograd():
            output = self.model(tensor)

        return {"prediction": output.cpu().tolist()}

2. TensorFlow Model

from ray import serve
import tensorflow as tf

@serve.deployment
class TensorFlowModel:
    def init(self, modelpath: str):

        self.model = tf.savedmodel.load(modelpath)


    async def call(self, request):
        data = await request.json()
        inputtensor = tf.constant(data["input"])
        output = self.model(inputtensor)

        return {"prediction": output.numpy().tolist()}

from ray import serve
from transformers import pipeline

@serve.deployment(rayactoroptions={"numgpus": 1})
class TransformersModel:
    def init(self):
        self.classifier = pipeline(
            "sentiment-analysis",
            model="distilbert-base-uncased-finetuned-sst-2-english",
            device=0
        )

    async def call(self, request):
        data = await request.json()
        text = data["text"]
        result = self.classifier(text)
        return {"prediction": result}

serve.run(TransformersModel.bind())

Production Features

1. Health Checks

from ray import serve
from fastapi import FastAPI

app = FastAPI()

@serve.deployment(
    healthcheckperiods=10,

    healthchecktimeouts=30
)
@serve.ingress(app)
class ProductionModel:
    def init(self):
        self.model = self.loadmodel()
        self.healthy = True

    def loadmodel(self):
        return "productionmodel"


    def checkhealth(self):
        return self.healthy

    @app.get("/predict")
    async def predict(self, text: str):
        if not self.healthy:
            return {"error": "Model unhealthy"}
        return {"prediction": f"Result for {text}"}

    @app.get("/health")
    async def health(self):
        return {"healthy": self.healthy}

2. Logging and Monitoring

from ray import serve
import logging
import time

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("ray.serve")

@serve.deployment
class MonitoredModel:
    def init(self):
        self.model = "monitoredmodel"

        self.requestcount = 0

    async def call(self, request):
        starttime = time.time()

        self.requestcount += 1

        try:
            data = await request.json()
            result = {"prediction": "result"}

            latency = time.time() - starttime

            logger.info(f"Request {self.requestcount}: latency={latency:.3f}s")

            return result
        except Exception as e:
            logger.error(f"Error processing request: {e}")
            raise

3. Graceful Shutdown

from ray import serve
import signal

@serve.deployment(
    gracefulshutdowntimeouts=30,

    gracefulshutdownwaitloops=2

)
class GracefulModel:
    def init(self):
        self.model = "gracefulmodel"
        self.shuttingdown = False


    def reconfigure(self, config):
        if config.get("shutdown"):
            self.shuttingdown = True

    async def call(self, request):
        if self.shuttingdown:

            return {"error": "Server shutting down"}
        return {"result": "processed"}

from ray import serve

Version 1 (Blue)
@serve.deployment(name="modelv1")
class ModelV1:
    def call(self, request):
        return {"version": "v1", "result": "blue"}

Version 2 (Green)
@serve.deployment(name="modelv2")

class ModelV2:
    def call(self, request):
        return {"version": "v2", "result": "green"}

Router for traffic splitting
@serve.deployment(routeprefix="/model")
class TrafficRouter:
    def init(self, v1, v2, v2percentage: float = 0.0):

        self.v1 = v1
        self.v2 = v2
        self.v2percentage = v2percentage


    async def call(self, request):
        import random
        if random.random() < self.v2percentage:
            return await self.v2.remote(request)
        return await self.v1.remote(request)

Deploy with 20% traffic to v2
v1 = ModelV1.bind()
v2 = ModelV2.bind()
router = TrafficRouter.bind(v1, v2, v2percentage=0.2)

serve.run(router)

from ray import serve
import time

@serve.deployment
class CanaryRouter:
    def init(self, stable, canary):
        self.stable = stable
        self.canary = canary
        self.canarystarttime = time.time()

        self.canaryduration = 3600  # 1 hour

    async def call(self, request):
        elapsed = time.time() - self.canarystarttime
        canarypercentage = min(elapsed / self.canaryduration, 1.0) * 0.5

        import random
        if random.random() < canarypercentage:

            return await self.canary.remote(request)
        return await self.stable.remote(request)

# serveconfig.yaml
proxylocation: EveryNode


httpoptions:
  host: 0.0.0.0
  port: 8000

applications:
name: mlservice

    routeprefix: /ml

    importpath: app:deployment

    runtimeenv:
      pip:
torch
transformers
    deployments:

name: Model
        numreplicas: 3

        rayactoroptions:

          numcpus: 2
          numgpus: 1

# Deploy from config
serve deploy serveconfig.yaml

2. Programmatic Config

from ray import serve
from ray.serve.config import HTTPOptions, ProxyLocation

serve.start(
    httpoptions=HTTPOptions(

        host="0.0.0.0",
        port=8000,
        rootpath="/api"
    ),
    proxylocation=ProxyLocation.EveryNode

)

from ray import serve
import torch

@serve.deployment
class EfficientModel:
    def init(self, modelpath: str):
        # Load model once at initialization
        self.model = self.loadmodel(modelpath)


    def loadmodel(self, path):

        model = torch.load(path, maplocation="cpu")
        model.eval()
        return model

    def reconfigure(self, config):
        # Hot reload model without restart
        if "modelpath" in config:

            self.model = self.loadmodel(config["modelpath"])

    async def call(self, request):
        data = await request.json()
        with torch.nograd():

            return {"result": self.model(data["input"]).tolist()}

from ray import serve
from fastapi import FastAPI, HTTPException
import logging

app = FastAPI()
logger = logging.getLogger(name)

@serve.deployment
@serve.ingress(app)
class RobustModel:
    def init(self):
        self.model = self.loadmodel()


    def loadmodel(self):

        return "model"

    @app.post("/predict")
    async def predict(self, data: dict):
        try:
            if "input" not in data:
                raise HTTPException(statuscode=400, detail="Missing input")

            result = self.inference(data["input"])

            return {"prediction": result}

        except ValueError as e:
            logger.error(f"Validation error: {e}")
            raise HTTPException(statuscode=400, detail=str(e))

        except Exception as e:
            logger.error(f"Inference error: {e}")
            raise HTTPException(statuscode=500, detail="Internal error")


    def inference(self, input_data):
        return "result"

Conclusion

Ray Serve is essential for production ML serving with:

Key takeaways:

• Use autoscaling for dynamic workloads
• Enable batching for throughput optimization
• Compose models for complex pipelines
• Implement proper health checks
• Use deployment strategies for safe rollouts