Home Knowledge Base Health Checks for ML Services

Health Checks for ML Services

Types of Health Checks

CheckPurposeKubernetes
LivenessIs the process alive?livenessProbe
ReadinessCan it accept traffic?readinessProbe
StartupHas it fully started?startupProbe

Implementation

Basic Health Endpoint

from fastapi import FastAPI

app = FastAPI()

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

@app.get("/ready")
def ready():
    # Check dependencies
    if not model_loaded:
        return {"status": "not ready"}, 503
    if not can_connect_to_db():
        return {"status": "not ready"}, 503
    return {"status": "ready"}

Model Health Check

@app.get("/health/model")
async def model_health():
    try:
        # Run inference on test input
        start = time.time()
        result = model.predict(test_input)
        latency = time.time() - start

        return {
            "status": "healthy",
            "model_loaded": True,
            "inference_latency_ms": latency * 1000
        }
    except Exception as e:
        return {"status": "unhealthy", "error": str(e)}, 503

Kubernetes Configuration

apiVersion: apps/v1
kind: Deployment
spec:
  template:
    spec:
      containers:
      - name: llm-server
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
          failureThreshold: 3

        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 60  # Model loading time
          periodSeconds: 5
          failureThreshold: 2

        startupProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 0
          periodSeconds: 10
          failureThreshold: 30  # 5 minutes to start

ML-Specific Considerations

CheckWhat to Verify
Model loadedModel weights in memory
GPU availableCUDA device accessible
Warm-up completeFirst inference done
DependenciesVector DB, Redis connected

Deep Health Check

@app.get("/health/deep")
async def deep_health():
    checks = {
        "model": await check_model_health(),
        "gpu": await check_gpu_health(),
        "vector_db": await check_vector_db(),
        "cache": await check_redis(),
    }

    all_healthy = all(c["healthy"] for c in checks.values())
    status_code = 200 if all_healthy else 503

    return checks, status_code

Best Practices

health checklivenessreadiness

Explore 500+ Semiconductor & AI Topics

From EUV lithography to CUDA optimization — search the full knowledge base or chat with our AI assistant.