pip install -r requirements.txt
uvicorn src.serving_app:create_app --factory --reload --port 8000FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY src/ src/
COPY configs/ configs/
EXPOSE 8000
CMD ["uvicorn", "src.serving_app:create_app", "--factory", "--host", "0.0.0.0", "--port", "8000"]Build and run:
docker build -t model-server .
docker run -p 8000:8000 -v $(pwd)/models:/app/models model-serverapiVersion: apps/v1
kind: Deployment
metadata:
name: model-server
spec:
replicas: 3
selector:
matchLabels:
app: model-server
template:
metadata:
labels:
app: model-server
spec:
containers:
- name: server
image: your-registry.example.com/model-server:latest
ports:
- containerPort: 8000
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 10
periodSeconds: 15
readinessProbe:
httpGet:
path: /ready
port: 8000
initialDelaySeconds: 5
periodSeconds: 10
resources:
requests:
memory: "512Mi"
cpu: "500m"
limits:
memory: "2Gi"
cpu: "2"curl -X POST http://localhost:8000/predict \
-H "Content-Type: application/json" \
-d '{"features": {"age": 35, "income": 55000, "region": "north"}}'curl -X POST http://localhost:8000/predict/batch \
-H "Content-Type: application/json" \
-d '{"instances": [{"age": 35, "income": 55000}, {"age": 42, "income": 72000}]}'curl -X POST http://localhost:8000/predict \
-H "Content-Type: application/json" \
-H "X-Model-Version: canary_model" \
-d '{"features": {"age": 35, "income": 55000}}'Increase replicas in Kubernetes:
kubectl scale deployment model-server --replicas=5For GPU models, request GPU resources:
resources:
limits:
nvidia.com/gpu: 1| Strategy | Latency | Throughput | When to Use |
|---|---|---|---|
| No batching | Lowest | Lowest | Real-time APIs, <10ms SLA |
| Client-side batching | Medium | High | Bulk scoring, nightly jobs |
| Server-side adaptive | Medium | Highest | GPU models, high traffic |
For server-side adaptive batching, use BentoML or Triton (see configs/).
Key metrics to track:
For models with repeated inputs (e.g. recommendation), add a cache:
from functools import lru_cache
@lru_cache(maxsize=10000)
def cached_predict(features_tuple):
return model.predict(features_tuple)Convert dicts to tuples for hashability.