MLOps with GPU: From CI/CD to Model Serving

"Infrastructure as code, models as artifacts, experiments as science."

August 2026 · Giacomo Saccaggi

---

The MLOps Loop

Machine learning in production is fundamentally different from machine learning in notebooks. A model that achieves 95% accuracy in development means nothing if it can't be reliably deployed, monitored, and updated. This is where MLOps comes in — the discipline of applying DevOps principles to machine learning systems.

The MLOps lifecycle forms a continuous loop:

Why Automation Matters for ML

Traditional software is deterministic — given the same inputs, it produces the same outputs. Machine learning is different. Models are statistical artifacts that can degrade over time, even without code changes:

• Reproducibility: Without version control for data, code, and hyperparameters, you can't recreate past results
• Model decay: The world changes. A model trained on 2024 data may perform poorly on 2026 data
• Data drift: Input distributions shift. What was normal yesterday may be an outlier today
• Concept drift: The relationship between inputs and outputs changes over time

Key Principles of MLOps

Principle	Description	Implementation
Version Control	Track code, data, models, and experiments	Git, DVC, MLflow
Automated Testing	Validate code quality and model behavior	pytest, unit tests, integration tests
CI/CD	Continuous integration and deployment	GitHub Actions, Jenkins, GitLab CI
Containerization	Reproducible environments	Docker, Kubernetes
Monitoring	Track model performance in production	Prometheus, Grafana, custom dashboards

---

Project Structure and the Makefile Approach

A well-organized ML project separates concerns: data processing, model training, serving, and infrastructure. But how do you coordinate these components? The answer is surprisingly old-school: make.

Makefiles provide a declarative way to define tasks and their dependencies. They're language-agnostic, well-understood, and integrate seamlessly with CI/CD systems.

The Makefile

Here's a production-ready Makefile for an ML project with GPU support:

# Makefile for ML project with GPU support

install:
	pip install --upgrade pip && pip install -r requirements.txt
	pip install --upgrade --no-deps --force-reinstall git+https://github.com/openai/whisper.git

test:
	python -m pytest -vv --cov=main --cov=mylib test_*.py

format:
	black *.py hugging-face/zero_shot_classification.py hugging-face/hf_whisper.py

lint:
	pylint --disable=R,C --ignore-patterns=test_.*?py *.py mylib/*.py

container-lint:
	docker run --rm -i hadolint/hadolint < Dockerfile

checkgpu:
	echo "Checking GPU for PyTorch"
	python utils/verify_pytorch.py
	echo "Checking GPU for Tensorflow"
	python utils/verify_tf.py

refactor: format lint

deploy:
	#deploy goes here

all: install lint test format deploy

Understanding Each Target

Target	Purpose	When to Use
install	Install all dependencies including Whisper from source	Initial setup, dependency updates
test	Run pytest with coverage reporting	Before commits, in CI
format	Auto-format code with Black	Before commits
lint	Static analysis with pylint (refactoring and convention warnings disabled)	Before commits, in CI
container-lint	Validate Dockerfile best practices with Hadolint	When modifying Dockerfiles
checkgpu	Verify GPU availability for both PyTorch and TensorFlow	On GPU machines, debugging
refactor	Combined format + lint	Quick cleanup
deploy	Deployment placeholder	Production releases
all	Full pipeline: install → lint → test → format → deploy	CI/CD, full validation

Why Disable pylint R and C?

The flags --disable=R,C turn off Refactor and Convention warnings. This is intentional:

• R (Refactor): Suggestions like "too many arguments" or "too many local variables" — often false positives in data science code
• C (Convention): Style issues like "missing docstring" — handled by Black and team conventions instead

This keeps pylint focused on actual errors (E) and warnings (W) that indicate bugs or problematic patterns.

---

CI/CD with GitHub Actions

Continuous Integration ensures that every code change is automatically validated. GitHub Actions provides a powerful, free CI/CD platform that integrates directly with your repository.

The Workflow File

Create .github/workflows/ci.yml:

name: CI
on:
  push:
    branches: [ "GPU" ]
  pull_request:
    branches: [ "GPU" ]
  workflow_dispatch:

jobs:
  build:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - name: install packages
        run: make install
      - name: lint
        run: make lint
      - name: test
        run: make test
      - name: format
        run: make format
      - name: deploy
        run: make deploy

Understanding the Workflow

Triggers (on:)

• push: branches: ["GPU"] — Run on every push to the GPU branch
• pull_request: branches: ["GPU"] — Run on PRs targeting the GPU branch
• workflow_dispatch: — Allow manual triggering from the GitHub UI

The Mirror Principle

Notice how the CI steps mirror the Makefile targets exactly. This is intentional:

Why This Ensures Code Quality

1. Linting catches bugs early: Undefined variables, unused imports, and type mismatches are caught before they reach production
2. Tests validate behavior: Automated tests ensure that code changes don't break existing functionality
3. Formatting ensures consistency: Black enforces a consistent code style across the team
4. Blocking merges on failure: Configure branch protection rules to require passing CI before merging

---

GPU Verification and Multi-Framework Support

Modern ML workloads demand GPU acceleration. Training a transformer model on CPU might take days; on a GPU, it takes hours. But GPU environments are notoriously finicky — driver versions, CUDA versions, and framework versions must all align.

PyTorch GPU Verification

Create utils/verify_pytorch.py:

import torch

def verify_pytorch_gpu():
    """Verify PyTorch can access the GPU."""
    print(f"PyTorch version: {torch.__version__}")
    print(f"CUDA available: {torch.cuda.is_available()}")
    
    if torch.cuda.is_available():
        print(f"CUDA version: {torch.version.cuda}")
        print(f"cuDNN version: {torch.backends.cudnn.version()}")
        print(f"GPU count: {torch.cuda.device_count()}")
        
        for i in range(torch.cuda.device_count()):
            print(f"  GPU {i}: {torch.cuda.get_device_name(i)}")
            props = torch.cuda.get_device_properties(i)
            print(f"    Memory: {props.total_memory / 1e9:.1f} GB")
            print(f"    Compute capability: {props.major}.{props.minor}")
        
        # Quick tensor operation test
        x = torch.randn(1000, 1000, device="cuda")
        y = torch.matmul(x, x)
        print("✓ GPU tensor operation successful")
    else:
        print("✗ No GPU available for PyTorch")

if __name__ == "__main__":
    verify_pytorch_gpu()

Example output on a GPU machine:

# PyTorch version: 2.1.0+cu121
# CUDA available: True
# CUDA version: 12.1
# cuDNN version: 8902
# GPU count: 1
#   GPU 0: NVIDIA GeForce RTX 4090
#     Memory: 24.0 GB
#     Compute capability: 8.9
# ✓ GPU tensor operation successful

TensorFlow GPU Verification

Create utils/verify_tf.py:

import tensorflow as tf

def verify_tensorflow_gpu():
    """Verify TensorFlow can access the GPU."""
    print(f"TensorFlow version: {tf.__version__}")
    
    gpus = tf.config.list_physical_devices('GPU')
    print(f"GPUs available: {len(gpus)}")
    
    if gpus:
        for gpu in gpus:
            print(f"  {gpu.name}")
        
        # Quick tensor operation test
        with tf.device('/GPU:0'):
            x = tf.random.normal([1000, 1000])
            y = tf.matmul(x, x)
        print("✓ GPU tensor operation successful")
    else:
        print("✗ No GPU available for TensorFlow")
    
    # Show build info
    print(f"Built with CUDA: {tf.test.is_built_with_cuda()}")

if __name__ == "__main__":
    verify_tensorflow_gpu()

Running GPU Workloads in Containers

Docker containers can access host GPUs using the NVIDIA Container Toolkit:

# Install NVIDIA Container Toolkit (on host)
distribution=$(. /etc/os-release;echo $ID$VERSION_ID)
curl -s -L https://nvidia.github.io/nvidia-docker/gpgkey | sudo apt-key add -
curl -s -L https://nvidia.github.io/nvidia-docker/$distribution/nvidia-docker.list | \
    sudo tee /etc/apt/sources.list.d/nvidia-docker.list
sudo apt-get update && sudo apt-get install -y nvidia-container-toolkit
sudo systemctl restart docker

# Run container with GPU access
docker run --gpus all nvidia/cuda:12.1-base nvidia-smi

# Run with specific GPUs
docker run --gpus '"device=0,1"' my-ml-image python train.py

# Run with GPU memory limits
docker run --gpus all --memory=16g my-ml-image python train.py

Monitoring with nvidia-smi

The nvidia-smi command is essential for monitoring GPU utilization:

# One-time snapshot
nvidia-smi

# Continuous monitoring (every 1 second)
nvidia-smi -l 1

# Query specific metrics
nvidia-smi --query-gpu=name,memory.used,memory.total,utilization.gpu --format=csv

# Watch GPU memory during training
watch -n 1 nvidia-smi

Example output:

+-----------------------------------------------------------------------------+
| NVIDIA-SMI 535.104.05   Driver Version: 535.104.05   CUDA Version: 12.2     |
|-------------------------------+----------------------+----------------------+
| GPU  Name        Persistence-M| Bus-Id        Disp.A | Volatile Uncorr. ECC |
| Fan  Temp  Perf  Pwr:Usage/Cap|         Memory-Usage | GPU-Util  Compute M. |
|===============================+======================+======================|
|   0  NVIDIA GeForce ...  Off  | 00000000:01:00.0 Off |                  N/A |
| 30%   45C    P2    85W / 350W |   8192MiB / 24576MiB |     78%      Default |
+-------------------------------+----------------------+----------------------+

Why GPU Acceleration Matters

Task	CPU Time	GPU Time	Speedup
BERT Fine-tuning (1 epoch)	~4 hours	~15 minutes	16x
ResNet-50 Training (ImageNet)	~2 weeks	~8 hours	42x
Whisper Transcription (1hr audio)	~45 minutes	~3 minutes	15x
LLM Inference (GPT-2)	~200ms/token	~5ms/token	40x

---

Containerization with Docker

Containers solve the "it works on my machine" problem. By packaging your application with its dependencies, you ensure consistent behavior across development, testing, and production environments.

Basic Dockerfile

A minimal Dockerfile for a utility script:

FROM alpine:latest
RUN apk update && apk add bash
WORKDIR /app
COPY repeat.sh /app

This Dockerfile:

• Uses Alpine Linux as base (minimal ~5MB image)
• Installs bash (Alpine uses ash by default)
• Sets the working directory
• Copies a script into the container

ML-Optimized Dockerfile

For ML workloads, you need Python, CUDA, and your dependencies:

# Multi-stage build for smaller final image
FROM nvidia/cuda:12.1-cudnn8-devel-ubuntu22.04 AS builder

# Install Python and build dependencies
RUN apt-get update && apt-get install -y \
    python3.10 \
    python3-pip \
    python3.10-venv \
    git \
    && rm -rf /var/lib/apt/lists/*

WORKDIR /app

# Create virtual environment and install dependencies
COPY requirements.txt .
RUN python3 -m venv /opt/venv && \
    /opt/venv/bin/pip install --upgrade pip && \
    /opt/venv/bin/pip install -r requirements.txt

# Runtime stage - smaller base image
FROM nvidia/cuda:12.1-cudnn8-runtime-ubuntu22.04

RUN apt-get update && apt-get install -y \
    python3.10 \
    && rm -rf /var/lib/apt/lists/*

# Copy virtual environment from builder
COPY --from=builder /opt/venv /opt/venv

WORKDIR /app
COPY . .

ENV PATH="/opt/venv/bin:$PATH"
ENV PYTHONUNBUFFERED=1

CMD ["python3", "main.py"]

Development Containers with VS Code

For local development, VS Code's Dev Containers feature provides a consistent environment. Create .devcontainer/devcontainer.json:

{
    "name": "MLOps GPU Development",
    "image": "mcr.microsoft.com/devcontainers/python:3.10",
    "features": {
        "ghcr.io/devcontainers/features/docker-in-docker:2": {},
        "ghcr.io/devcontainers/features/github-cli:1": {}
    },
    "customizations": {
        "vscode": {
            "extensions": [
                "ms-python.python",
                "ms-python.vscode-pylance",
                "ms-azuretools.vscode-docker",
                "GitHub.copilot"
            ],
            "settings": {
                "python.defaultInterpreterPath": "/usr/local/bin/python",
                "python.linting.pylintEnabled": true,
                "python.formatting.provider": "black"
            }
        }
    },
    "postCreateCommand": "pip install -r requirements.txt",
    "forwardPorts": [8000, 8888],
    "runArgs": ["--gpus", "all"]
}

This configuration:

• Uses Microsoft's official Python dev container
• Adds Docker-in-Docker for building containers inside the dev container
• Installs useful VS Code extensions
• Configures Python tooling
• Forwards ports for web servers and Jupyter
• Enables GPU access

Multi-Stage Build Benefits

Stage	Purpose	Contents
Builder	Compile code, install dependencies	Build tools, compilers, dev headers
Runtime	Run the application	Only runtime libraries and your code

---

Model Serving with BentoML

Training a model is only half the battle. Serving it in production requires handling HTTP requests, batching predictions, managing model versions, and scaling under load. BentoML provides an elegant solution.

Saving a Model

First, train and save your model:

import bentoml
from sklearn import svm, datasets

# Load the famous Iris dataset
iris = datasets.load_iris()
X, y = iris.data, iris.target

# Train a Support Vector Classifier
clf = svm.SVC(gamma='scale')
clf.fit(X, y)

# Save to BentoML model store
saved_model = bentoml.sklearn.save_model("iris_clf", clf)
print(f"Model saved: {saved_model}")

# Output: Model saved: Model(tag="iris_clf:xyzabc123")

BentoML stores models in a local repository with automatic versioning. Each save creates a unique tag.

Defining a Service

Create service.py to define your API:

import numpy as np
import bentoml
from bentoml.io import NumpyNdarray

# Load the saved model
iris_clf_runner = bentoml.sklearn.get("iris_clf:latest").to_runner()

# Create the service
svc = bentoml.Service("iris_classifier", runners=[iris_clf_runner])

# Define the API endpoint
@svc.api(input=NumpyNdarray(), output=NumpyNdarray())
async def classify(input_array: np.ndarray) -> np.ndarray:
    """Classify iris samples.
    
    Args:
        input_array: Shape (n_samples, 4) with sepal/petal measurements
    
    Returns:
        Array of predicted class labels (0=setosa, 1=versicolor, 2=virginica)
    """
    return await iris_clf_runner.predict.async_run(input_array)

Running the Service

# Start the development server
bentoml serve service:svc --reload

# Server running at http://localhost:3000
# Swagger UI at http://localhost:3000/docs

Making Predictions

# Using curl
curl -X POST http://localhost:3000/classify \
    -H "Content-Type: application/json" \
    -d '[[5.1, 3.5, 1.4, 0.2], [6.7, 3.0, 5.2, 2.3]]'

# Response: [0, 2]  (setosa, virginica)

# Using Python requests
import requests

response = requests.post(
    "http://localhost:3000/classify",
    json=[[5.1, 3.5, 1.4, 0.2]]
)
print(response.json())  # [0]

Building and Deploying a Bento

Create bentofile.yaml:

service: "service:svc"
labels:
  owner: ml-team
  project: iris-classifier
include:
  - "*.py"
python:
  packages:
    - scikit-learn
    - numpy

# Build the Bento (containerizable artifact)
bentoml build

# Containerize it
bentoml containerize iris_classifier:latest

# Run the container
docker run -p 3000:3000 iris_classifier:latest

---

Zero-Shot Classification with HuggingFace

Traditional classification requires labeled training data for each category. What if you need to classify text into categories you've never seen during training? Zero-shot classification makes this possible.

The Implementation

from transformers import pipeline

def classify(text, labels, model="MoritzLaurer/mDeBERTa-v3-base-mnli-xnli"):
    """Classify text into any set of labels without training.
    
    Args:
        text: The text to classify
        labels: List of candidate labels
        model: HuggingFace model for zero-shot classification
    
    Returns:
        Classification results with scores for each label
    """
    classifier = pipeline("zero-shot-classification", model=model)
    results = classifier(text, labels, multi_label=False)
    return results

# Example usage
text = "The new iPhone features a faster processor and improved camera system."
labels = ["technology", "sports", "politics", "entertainment"]

result = classify(text, labels)
print(result)

Output:

# {
#     'sequence': 'The new iPhone features a faster processor and improved camera system.',
#     'labels': ['technology', 'entertainment', 'sports', 'politics'],
#     'scores': [0.92, 0.04, 0.02, 0.02]
# }

How Zero-Shot Classification Works

Zero-shot classification leverages Natural Language Inference (NLI) — a task where the model determines if a hypothesis follows from a premise.

When to Use Zero-Shot Classification

Use Case	Zero-Shot?	Why
Categories change frequently	✓ Yes	No retraining needed when labels change
Limited labeled data	✓ Yes	No training data required
Prototyping and exploration	✓ Yes	Quick experiments without data collection
High accuracy required	✗ Maybe	Fine-tuned models typically outperform
Fixed, well-defined categories	✗ Maybe	Consider fine-tuning for best performance

Multi-Label Classification

For texts that belong to multiple categories, set multi_label=True:

text = "The documentary explores both the political implications and the environmental impact of climate change."
labels = ["politics", "environment", "science", "entertainment"]

result = classifier(text, labels, multi_label=True)

# Scores are independent (can all be high):
# {'labels': ['environment', 'politics', 'science', 'entertainment'],
#  'scores': [0.94, 0.87, 0.72, 0.23]}

---

Speech-to-Text with Whisper

OpenAI's Whisper is a state-of-the-art speech recognition model trained on 680,000 hours of multilingual audio. It handles accents, background noise, and technical terminology remarkably well.

Basic Transcription

from transformers import pipeline

def transcribe(filename, model="openai/whisper-tiny.en"):
    """Transcribe audio file to text.
    
    Args:
        filename: Path to audio file (mp3, wav, flac, etc.)
        model: Whisper model variant
    
    Returns:
        Transcription results with text
    """
    pipe = pipeline("automatic-speech-recognition", model=model)
    results = pipe(filename)
    return results

# Example usage
result = transcribe("meeting_recording.mp3")
print(result["text"])

Whisper Model Variants

Model	Parameters	VRAM	Speed (RTF)	Best For
whisper-tiny	39M	~1GB	~32x	Quick prototypes, edge devices
whisper-base	74M	~1GB	~16x	Basic transcription
whisper-small	244M	~2GB	~6x	Good accuracy/speed balance
whisper-medium	769M	~5GB	~2x	High accuracy
whisper-large-v3	1550M	~10GB	~1x	Best accuracy, multilingual

RTF = Real-Time Factor. RTF of 10x means 1 hour of audio transcribed in 6 minutes.

Advanced Whisper Usage

from transformers import AutoModelForSpeechSeq2Seq, AutoProcessor, pipeline
import torch

def transcribe_advanced(filename, model_id="openai/whisper-large-v3"):
    """Advanced transcription with GPU acceleration and timestamps."""
    
    device = "cuda:0" if torch.cuda.is_available() else "cpu"
    torch_dtype = torch.float16 if torch.cuda.is_available() else torch.float32
    
    # Load model with optimizations
    model = AutoModelForSpeechSeq2Seq.from_pretrained(
        model_id,
        torch_dtype=torch_dtype,
        low_cpu_mem_usage=True,
        use_safetensors=True
    )
    model.to(device)
    
    processor = AutoProcessor.from_pretrained(model_id)
    
    pipe = pipeline(
        "automatic-speech-recognition",
        model=model,
        tokenizer=processor.tokenizer,
        feature_extractor=processor.feature_extractor,
        max_new_tokens=128,
        chunk_length_s=30,           # Process 30s chunks
        batch_size=16,               # Batch processing for speed
        return_timestamps=True,     # Include word timestamps
        torch_dtype=torch_dtype,
        device=device,
    )
    
    result = pipe(filename)
    return result

# Example with timestamps
result = transcribe_advanced("interview.mp3")
for chunk in result["chunks"]:
    start, end = chunk["timestamp"]
    print(f"[{start:.2f}s - {end:.2f}s]: {chunk['text']}")

---

Fine-Tuning with HuggingFace Trainer

While pre-trained models are powerful, fine-tuning on domain-specific data often yields significant improvements. HuggingFace's Trainer API abstracts away the training loop complexity.

Complete Fine-Tuning Example

from transformers import AutoTokenizer, AutoModelForSequenceClassification
from transformers import TrainingArguments, Trainer
from datasets import load_dataset
import numpy as np

# Load the Yelp reviews dataset
dataset = load_dataset("yelp_review_full")

# Load pre-trained tokenizer
tokenizer = AutoTokenizer.from_pretrained("bert-base-cased")

def tokenize_function(examples):
    """Tokenize text with padding and truncation."""
    return tokenizer(examples["text"], padding="max_length", truncation=True)

# Tokenize the entire dataset (batched for efficiency)
tokenized_datasets = dataset.map(tokenize_function, batched=True)

# Load pre-trained model with classification head
model = AutoModelForSequenceClassification.from_pretrained(
    "bert-base-cased", 
    num_labels=5  # 5 star ratings
)

# Create small subsets for demonstration
small_train = tokenized_datasets["train"].shuffle(seed=42).select(range(1000))
small_eval = tokenized_datasets["test"].shuffle(seed=42).select(range(1000))

# Define training arguments
training_args = TrainingArguments(
    output_dir="test_trainer",
    evaluation_strategy="epoch"
)

# Initialize Trainer
trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=small_train,
    eval_dataset=small_eval,
)

# Fine-tune the model
trainer.train()

Understanding the Components

Component	Purpose	Key Options
AutoTokenizer	Convert text to token IDs	padding, truncation, max_length
AutoModelForSequenceClassification	Pre-trained model + classification head	num_labels, problem_type
TrainingArguments	Training hyperparameters	learning_rate, batch_size, epochs
Trainer	Training loop abstraction	compute_metrics, callbacks

Evaluation Strategies

The evaluation_strategy parameter controls when validation runs:

• "no": No evaluation during training
• "epoch": Evaluate at the end of each epoch
• "steps": Evaluate every eval_steps steps

Custom Metrics

from sklearn.metrics import accuracy_score, f1_score
import numpy as np

def compute_metrics(eval_pred):
    """Compute accuracy and F1 score."""
    logits, labels = eval_pred
    predictions = np.argmax(logits, axis=-1)
    
    return {
        "accuracy": accuracy_score(labels, predictions),
        "f1_macro": f1_score(labels, predictions, average="macro"),
        "f1_weighted": f1_score(labels, predictions, average="weighted"),
    }

# Pass to Trainer
trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=small_train,
    eval_dataset=small_eval,
    compute_metrics=compute_metrics,
)

Production Training Arguments

training_args = TrainingArguments(
    output_dir="./results",
    
    # Training settings
    num_train_epochs=3,
    per_device_train_batch_size=16,
    per_device_eval_batch_size=64,
    
    # Optimizer settings
    learning_rate=2e-5,
    weight_decay=0.01,
    warmup_ratio=0.1,
    
    # Evaluation and saving
    evaluation_strategy="steps",
    eval_steps=500,
    save_strategy="steps",
    save_steps=500,
    save_total_limit=3,               # Keep only last 3 checkpoints
    load_best_model_at_end=True,
    metric_for_best_model="f1_macro",
    
    # GPU optimization
    fp16=True,                        # Mixed precision training
    dataloader_num_workers=4,
    
    # Logging
    logging_dir="./logs",
    logging_steps=100,
    report_to="tensorboard",
)

---

Testing in MLOps

Testing ML systems is challenging. Unlike traditional software where functions have deterministic outputs, ML models produce probabilistic results that can vary with random seeds, data order, and floating-point precision.

Unit Testing for ML Code

Create test_main.py:

from mylib.calculator import add

def test_add():
    """Test basic addition function."""
    assert add(1, 2) == 3

def test_add_negative():
    """Test addition with negative numbers."""
    assert add(-1, 1) == 0

def test_add_floats():
    """Test addition with floating point numbers."""
    result = add(0.1, 0.2)
    assert abs(result - 0.3) < 1e-9  # Float comparison with tolerance

The corresponding module mylib/calculator.py:

def add(a, b):
    """Add two numbers.
    
    Args:
        a: First number
        b: Second number
    
    Returns:
        Sum of a and b
    """
    return a + b

Running Tests with Coverage

# Run all tests with verbose output and coverage
python -m pytest -vv --cov=main --cov=mylib test_*.py

# Sample output:
# test_main.py::test_add PASSED                    [ 33%]
# test_main.py::test_add_negative PASSED           [ 66%]
# test_main.py::test_add_floats PASSED             [100%]
# 
# ---------- coverage: platform linux, python 3.10 ----------
# Name                  Stmts   Miss  Cover
# -----------------------------------------
# mylib/calculator.py       3      0   100%
# -----------------------------------------
# TOTAL                     3      0   100%

Testing ML-Specific Code

ML testing requires different strategies:

import pytest
import numpy as np
from unittest.mock import Mock, patch

class TestDataPipeline:
    """Tests for data preprocessing pipeline."""
    
    def test_normalize_output_range(self):
        """Normalized data should be in [0, 1] range."""
        from mylib.preprocessing import normalize
        
        data = np.array([10, 20, 30, 40, 50])
        result = normalize(data)
        
        assert result.min() >= 0
        assert result.max() <= 1
    
    def test_normalize_preserves_shape(self):
        """Normalization should not change array shape."""
        from mylib.preprocessing import normalize
        
        data = np.random.randn(100, 10)
        result = normalize(data)
        
        assert result.shape == data.shape
    
    def test_handles_empty_input(self):
        """Should handle empty arrays gracefully."""
        from mylib.preprocessing import normalize
        
        data = np.array([])
        with pytest.raises(ValueError):
            normalize(data)


class TestModelInference:
    """Tests for model inference (mocking the actual model)."""
    
    def test_predict_returns_correct_shape(self):
        """Predictions should have shape (n_samples, n_classes)."""
        from mylib.inference import predict
        
        # Mock the model to avoid loading actual weights
        mock_model = Mock()
        mock_model.predict.return_value = np.random.rand(10, 3)
        
        with patch('mylib.inference.load_model', return_value=mock_model):
            inputs = np.random.rand(10, 128)
            result = predict(inputs)
            
            assert result.shape == (10, 3)
    
    def test_predict_probabilities_sum_to_one(self):
        """Predicted probabilities should sum to 1 for each sample."""
        from mylib.inference import predict
        
        mock_probs = np.array([[0.7, 0.2, 0.1], [0.3, 0.3, 0.4]])
        mock_model = Mock()
        mock_model.predict.return_value = mock_probs
        
        with patch('mylib.inference.load_model', return_value=mock_model):
            inputs = np.random.rand(2, 128)
            result = predict(inputs)
            
            np.testing.assert_array_almost_equal(result.sum(axis=1), [1.0, 1.0])

Testing Data Pipelines vs Model Behavior

Test Type	What to Test	Example
Data Pipeline Tests	Input/output shapes, value ranges, handling edge cases	Normalization preserves shape, encoding handles unknown categories
Model Tests (Mocked)	Interface contracts, output format, error handling	Predictions have correct shape, API returns valid JSON
Integration Tests	End-to-end flow with real (but small) models	Load model, run inference, verify output format
Regression Tests	Model performance doesn't degrade	Accuracy on test set ≥ baseline threshold

---

Summary

We've covered the complete MLOps pipeline for GPU-accelerated machine learning:

Component	Tools	Purpose
Project Structure	Makefile	Reproducible commands for install, test, lint, deploy
CI/CD	GitHub Actions	Automated testing and deployment on every commit
GPU Support	PyTorch, TensorFlow, nvidia-smi	Hardware acceleration for training and inference
Containerization	Docker, Dev Containers	Reproducible environments across dev/test/prod
Model Serving	BentoML	Production-ready REST APIs for ML models
NLP Tasks	HuggingFace Transformers	Zero-shot classification, speech-to-text, fine-tuning
Testing	pytest	Unit tests, mocking, coverage reporting

Key Takeaways

The Complete Workflow

# 1. Development
git checkout -b feature/new-model
# Write code, train model...

# 2. Local validation
make lint          # Check code quality
make test          # Run tests
make checkgpu      # Verify GPU access

# 3. Commit and push
git add .
git commit -m "Add improved model architecture"
git push origin feature/new-model

# 4. CI/CD runs automatically
# GitHub Actions: install → lint → test → format → deploy

# 5. Model serving
bentoml serve service:svc                    # Local testing
bentoml build && bentoml containerize        # Production packaging
docker run --gpus all -p 3000:3000 my-model  # Deploy

---

References

• Sculley et al. (2015). "Hidden Technical Debt in Machine Learning Systems" — Google's seminal paper on ML infrastructure
• OpenAI Whisper — github.com/openai/whisper
• HuggingFace Transformers — huggingface.co/docs/transformers
• BentoML Documentation — docs.bentoml.org
• GitHub Actions Documentation — docs.github.com/en/actions
• NVIDIA Container Toolkit — docs.nvidia.com/datacenter/cloud-native/container-toolkit