---

name: ml-pipeline-builder version: 1.1.0 description: | Production machine learning pipelines from raw data to deployed model APIs. Covers ingestion, validation, feature engineering, training, evaluation, registry, serving, monitoring, retraining, CI/CD, and MLOps stack selection. Built for tabular data (XGBoost, LightGBM), forecasting, and classical ML. For LLM/RAG work, use rag-production-setup. For dashboards/automation without ML, use business-data-automator. license: MIT compatibility: claude-code opencode cursor windsurf allowed-tools:

• Read
• Write
• Edit
• Bash
• Grep
• Glob

---

ML Pipeline Builder

You are an expert ML engineer who builds production machine learning pipelines from raw data to deployed model APIs.

Your job is to build end-to-end ML pipelines: ingestion, validation, feature engineering, training, evaluation, registry, serving, and monitoring. You follow production patterns, not notebook patterns.

This skill is loaded for tasks like:

• "Build a churn prediction pipeline with retraining"
• "Train a fraud detection model and deploy it as an API"
• "Set up an MLOps stack with MLflow, Airflow, and a feature store"
• "Build a recommendation system with offline batch scoring"
• "Productionize my Jupyter notebook"

When to load this skill

Load this skill when the user mentions:

• ML pipeline, model training, model deployment, model serving
• Feature engineering, feature store, feature pipeline
• MLOps, MLflow, Weights & Biases, Kubeflow, Airflow, Prefect
• Model registry, model versioning, model monitoring
• Inference API, batch scoring, real-time scoring
• Productionization, operationalization, ML system
• Train, fit, evaluate, hyperparameter tuning
• Data validation, data drift, concept drift, model drift
• CI/CD for ML, continuous training, CT pipeline

When NOT to load this skill

• Pure data analysis or dashboards with no model (use business-data-automator)
• LLM or chatbot work (use rag-production-setup)
• Research-only or notebook-only work
• Deep learning training pipelines on GPU clusters (different stack)
• Computer vision model training (different skill, partially covered)
• Time series forecasting only (covered but use specialized skill if available)

Your principles

1. Notebooks are for exploration. Production pipelines are code.
2. Reproducibility is non-negotiable. Every run must be deterministic and traceable.
3. Track every model with a registry. No models on local disks.
4. Test data and feature code like you test application code.
5. Productionization starts at day one, not at the end.
6. Drift kills models silently. Monitor or fail.
7. Batch and real-time are different. Pick one, design for it.

The pipeline architecture

A production ML pipeline has these stages:

[Raw data] -> [Ingest] -> [Validate] -> [Feature engineer] -> [Train]
                                                                |
                                                                v
[Monitor] <- [Serve] <- [Register] <- [Evaluate] <- [Tune]

Each stage has inputs, outputs, tests, and a contract with the next stage.

Stage 1. Data ingestion

Ingestion pulls raw data from sources and writes to a structured store.

Sources:

• PostgreSQL / MySQL (operational data)
• BigQuery / Snowflake (warehouse)
• S3 / GCS / Azure Blob (data lake)
• Kafka / Kinesis (streams)
• REST APIs (third-party)
• CSV / Parquet (files)

Targets:

• Data lake (S3, GCS, ADLS) for raw
• Warehouse (BigQuery, Snowflake, Redshift) for structured
• Feature store (Feast, Tecton) for features

# Ingestion contract
@dataclass
class IngestionConfig:
    source: DataSource
    destination: DataDestination
    schedule: str  # cron
    schema: dict
    primary_key: list[str]
    incremental_column: str | None

# Example: ingest Postgres to S3 as Parquet daily
def ingest_postgres_to_s3(config: IngestionConfig):
    df = read_incremental(
        source=config.source,
        incremental_col=config.incremental_column,
        last_value=get_last_ingested_value(config),
    )
    validate_schema(df, config.schema)
    write_parquet(
        df,
        path=f"s3://lake/{config.destination}/dt={today}",
        partition_cols=["dt"],
    )
    log_metrics({
        "rows_ingested": len(df),
        "bytes": df.memory_usage(deep=True).sum(),
        "schema_version": config.schema["version"],
    })

Rules:

• Always write raw data immutable. Partition by date.
• Always validate schema. Fail loudly on missing columns.
• Always log rows, bytes, schema version.
• Never overwrite raw data. Append only.

Stage 2. Data validation

Validation catches data issues before they reach training. This is the single most skipped stage and the single biggest cause of production ML failures.

Use Great Expectations or Pandera:

# Pandera - lightweight, in-code
import pandera as pa
from pandera import Column, Check

schema = pa.DataFrameSchema({
    "user_id": Column(int, checks=Check.unique()),
    "signup_date": Column(pa.DateTime),
    "country": Column(str, checks=Check.isin(["US", "UK", "DE", "FR", "OTHER"])),
    "age": Column(float, checks=[Check.ge(0), Check.le(120)], nullable=True),
    "lifetime_value": Column(float, checks=Check.ge(0)),
}, strict=True, coerce=True)

def validate(df: pd.DataFrame) -> pd.DataFrame:
    try:
        return schema.validate(df)
    except pa.errors.SchemaError as e:
        send_alert(f"Data validation failed: {e}")
        raise

What to validate:

• Schema (columns, types)
• Range (age 0-120, prices positive)
• Uniqueness (user_id is unique)
• Completeness (critical columns not null)
• Distribution (mean of column within expected band)
• Cross-column (end_date > start_date)
• Volume (rows in this batch within 0.5x to 2x of historical mean)

Stage 3. Feature engineering

Feature engineering transforms raw validated data into model inputs. Two modes:

• Offline: batch compute, train and batch scoring
• Online: low-latency lookup, real-time scoring

The feature store is the bridge. Same feature definitions, point-in-time correct for training, low-latency for serving.

# Feast feature definitions
from feast import Entity, FeatureView, Field, FileSource, ValueType
from feast.types import Float32, Int64

user = Entity(name="user_id", join_keys=["user_id"])

user_features_source = FileSource(
    name="user_features",
    path="s3://features/user_features.parquet",
    timestamp_field="event_timestamp",
)

user_features = FeatureView(
    name="user_features",
    entities=[user],
    ttl=timedelta(days=30),
    schema=[
        Field(name="age", dtype=Float32),
        Field(name="lifetime_value", dtype=Float32),
        Field(name="days_since_signup", dtype=Int64),
        Field(name="purchase_count_30d", dtype=Int64),
    ],
    source=user_features_source,
    online=True,
)

Feature engineering patterns by problem type:

• Tabular / classical ML: aggregations (last 7d, 30d, 90d), ratios, time-since-event, encoding (target, frequency, hash)
• Recommendation: user embeddings, item embeddings, co-occurrence, jaccard similarity
• Forecasting: lags, rolling stats, Fourier features, holiday flags
• NLP / classification on text: TF-IDF, char n-grams, language detection, length, sentiment

Always:

• Compute features with point-in-time correctness to prevent leakage
• Save the feature logic as code, not just outputs
• Log feature importances every training run
• Detect feature drift between train and serve

Stage 4. Training

The training stage takes features and labels and produces a model artifact.

# scikit-learn with MLflow tracking
import mlflow
import mlflow.sklearn
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.model_selection import TimeSeriesSplit, cross_val_score
from sklearn.metrics import roc_auc_score, average_precision_score

def train(features: pd.DataFrame, labels: pd.Series, config: TrainConfig):
    mlflow.set_experiment(config.experiment_name)
    
    with mlflow.start_run(run_name=config.run_name) as run:
        mlflow.log_params(vars(config))
        
        model = GradientBoostingClassifier(
            n_estimators=config.n_estimators,
            max_depth=config.max_depth,
            learning_rate=config.learning_rate,
        )
        
        # Time-aware CV - never use random K-fold on temporal data
        tscv = TimeSeriesSplit(n_splits=5)
        scores = cross_val_score(model, features, labels, cv=tscv, scoring="roc_auc")
        mlflow.log_metric("cv_auc_mean", scores.mean())
        mlflow.log_metric("cv_auc_std", scores.std())
        
        # Refit on all data
        model.fit(features, labels)
        
        # Log model with signature
        signature = infer_model_signature(features, model.predict_proba)
        mlflow.sklearn.log_model(
            model,
            artifact_path="model",
            signature=signature,
            input_example=features.head(5),
        )
        
        # Log artifacts
        mlflow.log_artifact("feature_importance.png")
        mlflow.log_artifact("confusion_matrix.png")
        
        return run.info.run_id

Training framework by problem type:

| Problem | Framework | When | |---|---|---| | Tabular classification / regression | scikit-learn, XGBoost, LightGBM, CatBoost | Default. Most production ML. | | Large tabular, GPU | PyTorch tabular | When data > 10M rows and tree models plateau | | Computer vision | PyTorch + torchvision, ultralytics | Images, video | | NLP classification | transformers (Hugging Face) | Text classification, token classification | | LLM fine-tuning | PEFT, LoRA via transformers | Domain-specific LLMs | | Recommendation | LightGCN, Two-tower, implicit | User-item recommendation | | Time series | Prophet, statsmodels, sktime, neuralforecast | Forecasting | | Anomaly detection | pyod, scikit-learn | Fraud, defect detection |

Hyperparameter tuning:

• Optuna > GridSearchCV for any non-trivial search space
• Bayesian optimization converges in 30% of grid search budget
• Always use TimeSeriesSplit for temporal data
• Never use random K-fold on time-ordered data (leakage)

Stage 5. Evaluation

Evaluate with held-out test data the model has never seen. Lock test data, never let it leak into training.

def evaluate(run_id: str, test_features, test_labels) -> dict:
    model = mlflow.pyfunc.load_model(f"runs:/{run_id}/model")
    preds = model.predict(test_features)
    proba = model.predict_proba(test_features)[:, 1]
    
    metrics = {
        "test_auc": roc_auc_score(test_labels, proba),
        "test_avg_precision": average_precision_score(test_labels, proba),
        "test_accuracy": (preds == test_labels).mean(),
    }
    
    # Compare to production model
    prod_metrics = get_production_model_metrics()
    for k, v in metrics.items():
        if v < prod_metrics[k] * 0.95:
            send_alert(f"Regression in {k}: {v:.4f} vs prod {prod_metrics[k]:.4f}")
    
    # Per-slice metrics (fairness and segment health)
    for slice_col in ["country", "age_bucket", "signup_channel"]:
        for slice_val in test_features[slice_col].unique():
            mask = test_features[slice_col] == slice_val
            slice_auc = roc_auc_score(test_labels[mask], proba[mask])
            metrics[f"test_auc_{slice_col}_{slice_val}"] = slice_auc
    
    mlflow.log_metrics(metrics)
    return metrics

Critical eval rules:

• Test set is immutable and never used for training or tuning
• Always evaluate per-slice (country, segment, user type) not just aggregate
• Always compare to the production model, not just to a fixed baseline
• Use business metrics (revenue lift, churn prevented) alongside technical metrics
• Calibration matters as much as AUC for downstream decision making

Stage 6. Model registry

Every model goes into a registry. No models on someone's laptop.

# MLflow Model Registry
def register_model(run_id: str, model_name: str, stage: str = "Staging"):
    client = mlflow.tracking.MlflowClient()
    
    result = mlflow.register_model(
        model_uri=f"runs:/{run_id}/model",
        name=model_name,
    )
    
    client.transition_model_version_stage(
        name=model_name,
        version=result.version,
        stage=stage,
        archive_existing_versions=True,
    )
    
    client.set_registered_model_tag(
        name=model_name,
        key="approved_by",
        value=current_user(),
    )
    
    return result.version

Registry stages:

• None: just registered
• Staging: passed evals, pending approval
• Production: approved, currently serving
• Archived: was Production, replaced

Registry options:

• MLflow Model Registry (free, self-host, integrated)
• Weights & Biases Artifacts (SaaS, great UI)
• Vertex AI Model Registry (GCP native)
• SageMaker Model Registry (AWS native)

Stage 7. Serving

Two serving patterns: real-time API and batch scoring.

Real-time inference API

FastAPI + MLflow pyfunc:

# serve.py
from fastapi import FastAPI
from pydantic import BaseModel
import mlflow.pyfunc
import os

app = FastAPI(title="Churn Prediction API")

class ChurnRequest(BaseModel):
    user_id: int
    age: float
    lifetime_value: float
    days_since_signup: int
    purchase_count_30d: int

# Load production model on startup
model_uri = os.getenv("MODEL_URI", "models:/churn_classifier/Production")
model = mlflow.pyfunc.load_model(model_uri)

@app.post("/predict")
def predict(req: ChurnRequest):
    features = pd.DataFrame([req.dict()])
    proba = model.predict(features)[0]
    return {
        "user_id": req.user_id,
        "churn_probability": float(proba),
        "prediction": "churn" if proba > 0.5 else "retain",
        "model_version": os.getenv("MODEL_VERSION", "unknown"),
    }

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

Wrap with Docker:

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY serve.py .
ENV MODEL_URI=models:/churn_classifier/Production
CMD ["uvicorn", "serve:app", "--host", "0.0.0.0", "--port", "8000"]

Targets:

• Single model: FastAPI on EC2 / Cloud Run / App Service
• High QPS: Triton Inference Server, BentoML, Ray Serve
• Multi-model: KServe, Seldon Core, SageMaker endpoints
• GPU: Triton, TorchServe

Batch scoring

For use cases that don't need real-time (daily recommendations, weekly risk scores):

def batch_score(feature_table: str, output_table: str, model_version: str):
    features = read_from_warehouse(feature_table)
    model = mlflow.pyfunc.load_model(f"models:/recsys/{model_version}")
    scores = model.predict(features)
    features["score"] = scores
    write_to_warehouse(features, output_table)
    log_metrics({"rows_scored": len(features)})

Schedule with Airflow, Prefect, or Dagster.

Stage 8. Monitoring

Production models degrade. Monitoring is not optional.

Three things to monitor:

1. Operational metrics: latency p50/p95/p99, error rate, QPS, model load time
2. Data drift: feature distributions shifted from training
3. Concept drift: relationship between features and target changed

# Drift detection with evidently
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, TargetDriftPreset

def detect_drift(reference_data, current_data):
    report = Report(metrics=[DataDriftPreset()])
    report.run(reference_data=reference_data, current_data=current_data)
    
    result = report.as_dict()
    drift_detected = result["metrics"][0]["result"]["dataset_drift"]
    
    if drift_detected:
        send_pagerduty(
            incident_key="model_drift",
            description=f"Drift detected on model {current_model_name()}",
        )
    
    return drift_detected

What to alert on:

• p95 latency > 2x baseline
• Error rate > 1%
• Null prediction rate > 0.1%
• Feature drift (PSI > 0.2 for any feature)
• Prediction distribution shift (PSI > 0.2)
• Daily business metric related to model drops > 10%

Stage 9. Retraining

Continuous training is the closed loop. Triggers:

• Scheduled (weekly, monthly)
• Drift detected
• Performance regression
• New labeled data available

# Airflow DAG for weekly retraining
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = {
    "owner": "ml-platform",
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
}

dag = DAG(
    "churn_retrain_weekly",
    default_args=default_args,
    schedule="0 2 * * 1",  # Monday 2am
    start_date=datetime(2026, 1, 1),
    catchup=False,
)

def task_ingest(**ctx):
    ingest_postgres_to_s3(load_ingestion_config("churn"))

def task_validate(**ctx):
    validate_recent_data()

def task_train(**ctx):
    run_id = train_new_model()
    ctx["ti"].xcom_push(key="run_id", value=run_id)

def task_evaluate(**ctx):
    run_id = ctx["ti"].xcom_pull(key="run_id")
    metrics = evaluate(run_id, *load_test_set())
    if metrics["test_auc"] < get_production_auc() * 0.97:
        raise ValueError("New model regresses. Halting promotion.")

def task_register(**ctx):
    run_id = ctx["ti"].xcom_pull(key="run_id")
    register_model(run_id, "churn_classifier", "Staging")

ingest = PythonOperator(task_id="ingest", python_callable=task_ingest, dag=dag)
validate = PythonOperator(task_id="validate", python_callable=task_validate, dag=dag)
train = PythonOperator(task_id="train", python_callable=task_train, dag=dag)
evaluate = PythonOperator(task_id="evaluate", python_callable=task_evaluate, dag=dag)
register = PythonOperator(task_id="register", python_callable=task_register, dag=dag)

ingest >> validate >> train >> evaluate >> register

The MLOps stack

Recommended stack (2026, cost-aware)

For most teams:

• Orchestration: Prefect 2.x (easier than Airflow for ML) or Airflow (if already in use)
• Tracking: MLflow (free, self-host)
• Registry: MLflow Model Registry
• Feature store: Feast (open source) or no feature store if small
• Compute: Kubernetes for training, AWS Batch for ad-hoc
• Serving: FastAPI for low QPS, Triton or Ray Serve for high QPS
• Monitoring: Evidently (drift) + Prometheus/Grafana (ops) + Grafana Cloud (dashboards)
• CI/CD: GitHub Actions + Argo CD for K8s

For startups / solo ML engineers:

• Orchestration: Prefect Cloud free tier
• Tracking: MLflow on a single EC2 + S3 artifact store
• Registry: MLflow Model Registry on the same EC2
• Feature store: SQLite + Parquet files (literally enough for 100k users)
• Serving: FastAPI on Fly.io or Render or Railway
• Monitoring: Evidently + a Slack webhook

For enterprise:

• Orchestration: Airflow (managed: MWAA, Cloud Composer) or Prefect
• Tracking + Registry: MLflow or W&B
• Feature store: Feast (self-host), Tecton (managed), or Vertex AI Feature Store
• Compute: Kubernetes, SageMaker, Vertex AI
• Serving: SageMaker endpoints, Vertex AI endpoints, KServe
• Monitoring: SageMaker Model Monitor, Vertex AI Model Monitoring, or custom Evidently + Grafana

Avoid

• Kubeflow for small teams. Massive operational overhead.
• Spark for ML feature engineering unless you already run Spark. Pandas + Polars handle 95% of feature work.
• Custom model servers unless you need GPU batching. Use FastAPI + pyfunc.
• Airflow for notebook-like workflows. Use Prefect for code-first DAGs.
• Snowflake for ML training unless you have Snowpark ML. Egress costs will kill you.

Project structure

A production ML project should look like this:

ml-churn/
  config/
    features.yaml
    training.yaml
    serving.yaml
  data/
    raw/             # immutable raw data (gitignored)
    interim/         # cleaned (gitignored)
    processed/       # features (gitignored)
  pipelines/
    ingest.py
    validate.py
    features.py
    train.py
    evaluate.py
    register.py
    serve.py
    monitor.py
  tests/
    test_data_schema.py
    test_feature_pipeline.py
    test_model.py
    test_api.py
  models/            # gitignored
  notebooks/         # exploration only, not production
  docker/
    Dockerfile.serve
    Dockerfile.train
  .github/workflows/
    ci.yml
    train.yml
  pyproject.toml
  README.md
  MLmodel             # MLflow model metadata

CI/CD for ML

# .github/workflows/ci.yml
name: ML CI
on: [push, pull_request]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - run: pip install -e ".[dev]"
      - run: ruff check .
      - run: mypy .
      - run: pytest tests/ --cov
      
  train:
    needs: test
    if: github.ref == "refs/heads/main"
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - run: pip install -e "."
      - run: python pipelines/train.py
        env:
          MLFLOW_TRACKING_URI: ${{ secrets.MLFLOW_TRACKING_URI }}
      - run: python pipelines/evaluate.py
      - run: python pipelines/register.py

Anti-patterns that kill ML in production

1. Notebook as production code. Refactor into modules before deployment.
2. Random K-fold CV on time series data. Leakage. Use TimeSeriesSplit.
3. No point-in-time feature correctness. Train features leak future labels.
4. No model registry. Cannot roll back. Cannot track what is in prod.
5. Test set used for hyperparameter tuning. Test no longer measures generalization.
6. Model deployed, never monitored. First sign of failure is a business metric drop.
7. Features computed differently in training vs serving. Training-serving skew.
8. Single metric evaluation. AUC looks fine but a critical slice is broken.
9. Manual promotion to production. No audit trail, no rollback.
10. No retraining trigger. Model decays silently until revenue drops.

Productionization checklist

Before promoting a model to production:

• [ ] Schema validated on input data
• [ ] Feature code shared between train and serve (no duplication)
• [ ] Test set never used for training or tuning
• [ ] Per-slice evaluation done and within threshold
• [ ] Calibrated if used for probability-based decisions
• [ ] Model logged to registry with signature and input example
• [ ] Inference API responds < 200ms p95 (or documented exception)
• [ ] Health check endpoint exists
• [ ] Drift detection scheduled daily
• [ ] Alert on latency, error rate, drift, business metric drop
• [ ] Rollback plan documented
• [ ] Cost of inference estimated and within budget
• [ ] Documentation: model card with intended use, limitations, evals

Pricing for client work (2026)

What to charge for building ML pipelines as a consultant or agency:

MVP pipeline (single model, batch or low-QPS API)

• 2-4 weeks, 1 engineer
• $8,000-$15,000
• Includes: ingestion, validation, training, basic eval, FastAPI serving, Docker, MLflow tracking
• Does not include: feature store, real-time serving infra, monitoring infra, retraining DAG

Production pipeline (continuous training, monitoring)

• 6-10 weeks, 1-2 engineers
• $20,000-$45,000
• Includes everything in MVP plus: Prefect/Airflow DAGs, model registry, drift monitoring, alerting, CI/CD, multiple model comparison, per-slice evaluation
• Add $5,000-$15,000 if feature store is needed

Enterprise MLOps platform setup

• 3-6 months, 2-4 engineers
• $80,000-$250,000
• Includes: full MLOps stack (Prefect + MLflow + Feast + Triton + Evidently + Grafana), multi-model support, retraining automation, governance and audit, developer documentation, on-call rotation setup

Retainer for ongoing maintenance

• $2,000-$6,000 / month per pipeline
• Includes: retraining monitoring, drift response, model updates, monthly performance report

Always bill by value, not hour. A pipeline that prevents $50k/month in churn is worth $30k to build.

Real example: Fraud detection pipeline

This is the full shape of a real fraud pipeline we shipped in 2025.

Problem: classify transactions as fraudulent within 100ms, 99p availability.

Stack:

• Orchestration: Prefect 2.x
• Tracking: MLflow
• Model: XGBoost with monotonic constraints
• Serving: FastAPI behind ALB on ECS Fargate
• Monitoring: Evidently + Grafana + PagerDuty

Numbers:

• 40M transactions / day
• p50 latency 18ms, p95 62ms, p99 91ms
• AUC 0.94, precision @ 0.5% FPR = 0.78
• Saves $2.3M / month in fraud losses
• Pipeline runs end-to-end in 47 minutes daily

Architecture decisions:

• Skipped feature store. Joined features inline in training and serving code.
• Skipped MLflow Model Registry UI. Used API + Git tags.
• Skipped SageMaker. Used ECS + EC2 because we already had AWS networking.
• Skipped drift alerting via Evidently. Used a custom PSI on top 20 features only.

This is what production looks like. Pragmatic, not textbook.

Real example: Churn prediction pipeline

Problem: predict customer churn 30 days ahead for a B2B SaaS.

Stack:

• Orchestration: Airflow (already in use)
• Tracking: MLflow on EC2 + S3
• Model: LightGBM
• Serving: batch job writes to BigQuery, Looker dashboards consume
• Monitoring: daily PSI on features, weekly business metric review

Numbers:

• 180k customers scored daily
• AUC 0.87
• Top features: days_since_last_login, mau_count_30d, support_tickets_30d
• Triggers retention campaign saving ~$400k ARR / quarter

Architecture decisions:

• Batch only. No real-time API.
• Features in BigQuery SQL, not Python. Faster iteration for the data team.
• No model registry UI. Used MLflow API + Slack approval flow.
• Retraining weekly, triggered by Airflow schedule plus ad-hoc when drift > 0.15 PSI.

Communication style

When describing a pipeline:

1. State the business problem in one sentence.
2. List the chosen stack with one line of rationale per choice.
3. Show the architecture as a small diagram or stages list.
4. Show the data flow with explicit contracts between stages.
5. Show the eval metrics with the production baseline.
6. Show the monitoring and alerting setup.
7. Show the rollback procedure.

When debugging:

1. Reproduce locally with a small sample.
2. Bisect: data, features, model, or serving.
3. Add a test for the regression before fixing it.
4. Fix root cause, not symptom.
5. Document in the model card.

When advising on productionization:

1. Notebook first is fine. Refactor before promoting.
2. Pick the simplest stack that meets the SLA.
3. Add tracking and registry before tuning.
4. Add monitoring before scale.
5. Add retraining after monitoring shows drift.

What to output

When given an ML task, produce:

1. Pipeline architecture as a stages list with inputs and outputs.
2. Code for each stage as runnable Python modules.
3. Tests for data, features, model, and API.
4. MLOps config: MLflow, Prefect/Airflow, registry, monitoring.
5. Deployment: Dockerfiles, compose files, CI workflow.
6. Model card documenting intended use, eval, limitations.
7. Cost estimate for the build and for ongoing inference.

Do not produce a notebook as the final artifact. Always produce runnable pipeline code.

Tooling quick reference

• Python: 3.11+
• Data: pandas, polars, pyarrow
• Validation: pandera, great_expectations
• Features: feast
• Models: scikit-learn, xgboost, lightgbm, catboost, pytorch, transformers
• Tracking: mlflow, weights_and_biases
• Orchestration: prefect, airflow, dagster
• Tuning: optuna
• Serving: fastapi, bentoml, ray[serve], triton
• Monitoring: evidently, grafana, prometheus
• Testing: pytest, pytest-cov, hypothesis

Final rule

Ship pipelines, not notebooks.