# Model Serving

> When to activate: model serving, inference, FastAPI ML, BentoML, Triton, vLLM, model deployment, online inference, batching

- Skill: `mattakushi432/model-serving` (Agent Skill)
- Install (CLI): `npx skillmds@latest add mattakushi432/model-serving`
- Raw SKILL.md: https://api.skillmd.com/api/skills/mattakushi432/model-serving/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: AI & ML
- Author: Mattakushi432 (https://skillmd.com/u/mattakushi432)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/mattakushi432/model-serving

---

# Model Serving Patterns

## FastAPI Inference Endpoint

```python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import joblib
import numpy as np
from contextlib import asynccontextmanager

model = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global model
    model = joblib.load("model.pkl")
    yield
    model = None

app = FastAPI(lifespan=lifespan)

class PredictRequest(BaseModel):
    features: list[float]

class PredictResponse(BaseModel):
    prediction: float
    probability: float | None = None

@app.post("/predict", response_model=PredictResponse)
async def predict(req: PredictRequest):
    X = np.array(req.features).reshape(1, -1)
    pred = model.predict(X)[0]
    proba = None
    if hasattr(model, "predict_proba"):
        proba = float(model.predict_proba(X)[0, 1])
    return PredictResponse(prediction=float(pred), probability=proba)

@app.get("/health")
async def health():
    return {"status": "ok", "model_loaded": model is not None}
```

## Async Batching

```python
import asyncio
from collections import deque

class BatchInferenceServer:
    def __init__(self, model, batch_size=32, max_wait_ms=10):
        self.model = model
        self.batch_size = batch_size
        self.max_wait = max_wait_ms / 1000
        self.queue: deque = deque()

    async def predict(self, features: list[float]) -> float:
        future = asyncio.get_event_loop().create_future()
        self.queue.append((features, future))
        if len(self.queue) >= self.batch_size:
            await self._flush()
        else:
            await asyncio.sleep(self.max_wait)
            if not future.done():
                await self._flush()
        return await future

    async def _flush(self):
        batch, futures = [], []
        while self.queue and len(batch) < self.batch_size:
            feat, fut = self.queue.popleft()
            batch.append(feat)
            futures.append(fut)
        preds = self.model.predict(batch)
        for fut, pred in zip(futures, preds):
            if not fut.done():
                fut.set_result(float(pred))
```

## BentoML Service

```python
import bentoml
import numpy as np

# Save model to BentoML store
bentoml.sklearn.save_model("classifier", model)

runner = bentoml.sklearn.get("classifier:latest").to_runner()

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

@svc.api(input=bentoml.io.NumpyNdarray(), output=bentoml.io.NumpyNdarray())
async def predict(input_data: np.ndarray) -> np.ndarray:
    return await runner.predict.async_run(input_data)

# bentoml serve service:svc --reload
# bentoml build && bentoml containerize classifier_svc:latest
```

## vLLM for LLM Serving

```python
from vllm import LLM, SamplingParams

llm = LLM(
    model="meta-llama/Llama-3-8B-Instruct",
    tensor_parallel_size=2,
    gpu_memory_utilization=0.9,
    max_model_len=8192,
)

sampling_params = SamplingParams(
    temperature=0.7,
    top_p=0.9,
    max_tokens=512,
)

prompts = ["Explain RAII in C++:", "What is a monad?"]
outputs = llm.generate(prompts, sampling_params)

for output in outputs:
    print(output.outputs[0].text)

# OpenAI-compatible server:
# python -m vllm.entrypoints.openai.api_server --model meta-llama/Llama-3-8B-Instruct
```

## Triton Inference Server Config

```
# model_repository/classifier/config.pbtxt
name: "classifier"
platform: "sklearn"
max_batch_size: 64
input [{ name: "input__0" data_type: TYPE_FP32 dims: [10] }]
output [{ name: "output__0" data_type: TYPE_FP32 dims: [1] }]
dynamic_batching { preferred_batch_size: [8, 16, 32] max_queue_delay_microseconds: 5000 }
instance_group [{ count: 2 kind: KIND_CPU }]
```

## Deployment Patterns

| Pattern | When to Use |
|---|---|
| Shadow mode | Test new model on real traffic without affecting users |
| A/B test | Compare two models with traffic split (e.g. 90/10) |
| Canary | Gradually roll out to increasing % of users |
| Feature flags | Gate model version per user segment |

```python
import random

def route_request(user_id: str, features: list) -> float:
    # 10% canary traffic to new model
    if random.random() < 0.1:
        return new_model.predict([features])[0]
    return old_model.predict([features])[0]
```