Skip to content
BSWEN
Blog

How to Use ML Pipeline Skill in Claude Code for Production-Grade Machine Learning

Cowrie
Cowrie
Dev @ Bswen

Purpose

This post demonstrates how to use the ML Pipeline skill in Claude Code to build production-grade machine learning infrastructure. I show how to trigger the skill, common use cases, and practical examples of training pipelines, experiment tracking, and model deployment workflows.

Environment

  • Claude Code with claude-skills plugin
  • PyTorch 2.0+
  • Python 3.9+
  • MLflow (for experiment tracking examples)

What is ML Pipeline Skill?

The ML Pipeline skill is a specialist in the claude-skills ecosystem that activates when I work on machine learning infrastructure. It provides expert guidance on building end-to-end ML workflows including feature engineering, distributed training, experiment tracking, and model deployment.

When I mention keywords like “ML pipeline”, “Kubeflow”, “MLflow”, “experiment tracking”, or “hyperparameter tuning”, Claude loads this skill and provides production-grade patterns for ML systems.

The skill focuses on:

  • Building feature engineering pipelines and feature stores
  • Orchestrating training workflows with Kubeflow, Airflow, or custom systems
  • Implementing experiment tracking with MLflow, Weights & Biases, or Neptune
  • Creating automated hyperparameter tuning pipelines
  • Setting up model registries and versioning systems
  • Designing data validation and preprocessing workflows

How to Trigger ML Pipeline Skill

The skill activates automatically when I use specific ML-related keywords in my prompts. Here are common trigger phrases:

"Build an ML pipeline for model training"
"Set up MLflow experiment tracking"
"Create a distributed training pipeline with PyTorch"
"Implement hyperparameter tuning with Optuna"
"Design a feature store architecture"
"Set up Kubeflow pipelines for model training"
"Build a model validation and deployment workflow"

Example 1: Creating a Production Training Pipeline

When I asked Claude to “build a production-grade PyTorch training pipeline with MLflow tracking,” the ML Pipeline skill provided this complete training infrastructure:

import torch
import torch.nn as nn
from torch.utils.data import DataLoader, Dataset
from torch.optim import AdamW
from torch.optim.lr_scheduler import CosineAnnealingLR
import logging
from pathlib import Path
from dataclasses import dataclass
from typing import Optional
import json


logger = logging.getLogger(__name__)


@dataclass
class TrainingConfig:
    """Training hyperparameters and settings."""
    model_name: str
    batch_size: int = 32
    learning_rate: float = 1e-4
    weight_decay: float = 0.01
    epochs: int = 10
    warmup_steps: int = 100
    max_grad_norm: float = 1.0
    seed: int = 42
    checkpoint_dir: str = "./checkpoints"
    log_every_n_steps: int = 100
    eval_every_n_steps: int = 500
    save_every_n_steps: int = 1000
    mixed_precision: bool = True
    gradient_accumulation_steps: int = 1


class Trainer:
    """Production-grade PyTorch trainer."""


    def __init__(
        self,
        model: nn.Module,
        config: TrainingConfig,
        train_dataloader: DataLoader,
        eval_dataloader: Optional[DataLoader] = None,
        experiment_tracker=None,
    ):
        self.model = model
        self.config = config
        self.train_dataloader = train_dataloader
        self.eval_dataloader = eval_dataloader
        self.tracker = experiment_tracker


        self._setup_device()
        self._setup_training()
        self._setup_checkpointing()


    def _setup_device(self) -> None:
        """Configure device and move model."""
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.model = self.model.to(self.device)


        if self.config.mixed_precision and self.device.type == "cuda":
            self.scaler = torch.amp.GradScaler("cuda")
        else:
            self.scaler = None


        logger.info(f"Training on device: {self.device}")


    def _setup_training(self) -> None:
        """Initialize optimizer and scheduler."""
        self.optimizer = AdamW(
            self.model.parameters(),
            lr=self.config.learning_rate,
            weight_decay=self.config.weight_decay,
        )


        total_steps = len(self.train_dataloader) * self.config.epochs
        self.scheduler = CosineAnnealingLR(
            self.optimizer,
            T_max=total_steps,
            eta_min=self.config.learning_rate * 0.01,
        )


        self.global_step = 0
        self.best_eval_loss = float("inf")


    def train(self) -> dict:
        """Run training loop."""
        self._set_seed()
        self.model.train()


        for epoch in range(self.config.epochs):
            for batch_idx, batch in enumerate(self.train_dataloader):
                loss = self._training_step(batch)


                if self.global_step % self.config.log_every_n_steps == 0:
                    self._log_metrics({
                        "train/loss": loss,
                        "train/lr": self.scheduler.get_last_lr()[0],
                    })


                if self.global_step % self.config.eval_every_n_steps == 0:
                    eval_metrics = self.evaluate()
                    self._log_metrics(eval_metrics)


        return {
            "best_eval_loss": self.best_eval_loss,
            "final_train_loss": loss,
            "total_steps": self.global_step,
        }

The skill included:

  • Mixed precision training for faster computation
  • Gradient accumulation for larger effective batch sizes
  • Comprehensive logging to experiment tracking
  • Checkpointing with automatic best-model saving
  • Reproducibility through seed setting
  • Gradient clipping for training stability

Example 2: Distributed Training with Multiple GPUs

When I need to scale training across multiple GPUs, the ML Pipeline skill provides distributed training patterns:

import torch.distributed as dist
from torch.nn.parallel import DistributedDataParallel as DDP
from torch.utils.data.distributed import DistributedSampler
import os


def setup_distributed() -> tuple[int, int, int]:
    """Initialize distributed training environment."""
    if "RANK" in os.environ:
        rank = int(os.environ["RANK"])
        local_rank = int(os.environ["LOCAL_RANK"])
        world_size = int(os.environ["WORLD_SIZE"])
    else:
        rank = 0
        local_rank = 0
        world_size = 1


    if world_size > 1:
        dist.init_process_group(
            backend="nccl",
            init_method="env://",
            world_size=world_size,
            rank=rank,
        )
        torch.cuda.set_device(local_rank)


    return rank, local_rank, world_size


def create_distributed_dataloader(
    dataset: Dataset,
    batch_size: int,
    world_size: int,
    rank: int,
    shuffle: bool = True,
) -> DataLoader:
    """Create DataLoader with distributed sampler."""
    sampler = DistributedSampler(
        dataset,
        num_replicas=world_size,
        rank=rank,
        shuffle=shuffle,
    )


    return DataLoader(
        dataset,
        batch_size=batch_size,
        sampler=sampler,
        num_workers=4,
        pin_memory=True,
        drop_last=True,
    )

The skill ensures proper distributed training setup:

  • NCCL backend for GPU communication
  • DistributedSampler for even data distribution
  • Proper rank-based checkpointing (save on rank 0 only)
  • Synchronized logging across processes

Example 3: Hyperparameter Tuning with Optuna

When I asked to “implement hyperparameter optimization with Optuna,” the ML Pipeline skill provided this complete HPO setup:

import optuna
from optuna.trial import Trial


def create_objective(
    train_dataset: Dataset,
    eval_dataset: Dataset,
    model_class: type,
) -> callable:
    """Create Optuna objective function."""


    def objective(trial: Trial) -> float:
        # Sample hyperparameters
        config = TrainingConfig(
            model_name="tuned_model",
            learning_rate=trial.suggest_float("lr", 1e-5, 1e-3, log=True),
            batch_size=trial.suggest_categorical("batch_size", [16, 32, 64]),
            weight_decay=trial.suggest_float("weight_decay", 1e-5, 1e-2, log=True),
            epochs=trial.suggest_int("epochs", 3, 10),
        )


        # Create model and train
        model = model_class(
            hidden_size=trial.suggest_categorical("hidden_size", [128, 256, 512]),
            num_layers=trial.suggest_int("num_layers", 2, 6),
            dropout=trial.suggest_float("dropout", 0.1, 0.5),
        )


        trainer = Trainer(
            model=model,
            config=config,
            train_dataloader=DataLoader(train_dataset, batch_size=config.batch_size),
            eval_dataloader=DataLoader(eval_dataset, batch_size=config.batch_size),
        )


        # Report intermediate values for pruning
        for epoch in range(config.epochs):
            trainer.train_epoch()
            eval_loss = trainer.evaluate()["eval/loss"]


            trial.report(eval_loss, epoch)


            if trial.should_prune():
                raise optuna.TrialPruned()


        return trainer.best_eval_loss


    return objective


def run_hyperparameter_search(
    train_dataset: Dataset,
    eval_dataset: Dataset,
    model_class: type,
    n_trials: int = 100,
) -> optuna.Study:
    """Run hyperparameter optimization with Optuna."""


    pruner = optuna.pruners.MedianPruner(
        n_startup_trials=5,
        n_warmup_steps=3,
        interval_steps=1,
    )


    study = optuna.create_study(
        study_name="hpo_study",
        direction="minimize",
        pruner=pruner,
        storage=f"sqlite:///hpo_study.db",
        load_if_exists=True,
    )


    objective = create_objective(train_dataset, eval_dataset, model_class)


    study.optimize(
        objective,
        n_trials=n_trials,
        timeout=3600 * 12,  # 12 hours
        n_jobs=1,  # Sequential for GPU
        show_progress_bar=True,
    )


    return study

The skill includes:

  • Median pruner to stop poorly performing trials early
  • SQLite storage for experiment persistence
  • Intermediate value reporting for pruning
  • Proper search space definition (log-scale for learning rates)
  • Checkpointing integration for resume capability

Best Practices

DO

  • Version all data, code, and models explicitly
  • Implement reproducible training environments (pinned dependencies, seeds)
  • Log all hyperparameters and metrics to experiment tracking
  • Validate data quality before training (schema checks, distribution validation)
  • Use containerized environments for training jobs
  • Implement proper error handling and retry logic
  • Store artifacts in versioned object storage
  • Enable pipeline monitoring and alerting
  • Document pipeline dependencies and data lineage

DON’T

  • Run training without experiment tracking
  • Deploy models without validation metrics
  • Hardcode hyperparameters in training scripts
  • Skip data validation and quality checks
  • Use non-reproducible random states
  • Store credentials in pipeline code
  • Train on production data without proper access controls
  • Deploy models without versioning
  • Ignore pipeline failures silently

When to Use ML Pipeline Skill

Use this skill when:

  • Building feature engineering pipelines and feature stores
  • Orchestrating training workflows with Kubeflow, Airflow, or custom systems
  • Implementing experiment tracking with MLflow, Weights & Biases, or Neptune
  • Creating automated hyperparameter tuning pipelines
  • Setting up model registries and versioning systems
  • Designing data validation and preprocessing workflows
  • Implementing model evaluation and validation strategies
  • Building reproducible training environments
  • Automating model retraining and deployment pipelines

Summary

In this post, I demonstrated how to use the ML Pipeline skill in Claude Code to build production-grade machine learning infrastructure. The key point is that this skill provides comprehensive patterns for the complete ML lifecycle, from feature engineering through model deployment, with emphasis on reproducibility, scalability, and observability.

The skill integrates with related claude-skills capabilities:

  • DevOps Engineer - CI/CD for training pipelines
  • Kubernetes Specialist - K8s-based training infrastructure
  • Cloud Architect - Cloud-native ML system design

Final Words + More Resources

My intention with this article was to help others share my knowledge and experience. If you want to contact me, you can contact by email: Email me

Here are also the most important links from this article along with some further resources that will help you in this scope:

Oh, and if you found these resources useful, don’t forget to support me by starring the repo on GitHub!

Comments

Tags: