pipeline-builder

Files (1)

• SKILL.md

SKILL.md (first lines)

# Pipeline Builder

Create automated, reliable ML pipelines with proper error handling, data validation, and scheduling.

---

## Workflow

```
1. Identify pipeline type and components
2. Select orchestration tool
3. Generate pipeline code
4. Add data quality gates and error handling
5. Configure scheduling and monitoring
```

---

## Step 1 -- Pipeline Type

| Type | Components | When |
|---|---|---|
| **Training pipeline** | Ingest -> Clean -> Feature Eng -> Train -> Evaluate -> Register | Scheduled retraining |
| **Inference pipeline** | Ingest -> Preprocess -> Predict -> Store results | Batch scoring |
| **Data pipeline** | Extract -> Transform -> Load -> Validate | Data preparation |
| **CI/CD for ML** | Test -> Train -> Evaluate -> Gate -> Deploy | Model deployment |
| **Monitoring pipeline** | Collect metrics -> Detect drift -> Alert -> Retrigger training | Model monitoring |

---

## Step 2 -- Select Orchestration Tool

| Tool | Best for | Complexity | When |
|---|---|---|---|
| **Simple Python script + cron** | Basic automation | Minimal | 1-3 steps, single machine |
| **Prefect** | Modern, Pythonic orchestration | Low | Default recommendation |
| **Airflow** | Enterprise standard, DAGs | Medium | Team environment, complex DAGs |
| **GitHub Actions** | CI/CD for ML, triggered by code change | Low | Model training on push |
| **Make / Just** | Local development pipelines | Minimal | Local reproducibility |

**Default**: Prefect for Python-native pipelines. Simple script + cron for quick solutions.

---

## Step 3 -- Pipeline Code

### Simple Script Pipeline

```python
"""ML training pipeline -- runs end-to-end."""
import logging
from datetime import datetime
from pathlib import Path

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

def step_ingest(config: dict) -> str:
    """Step 1: Load and validate raw data."""
    logger.info("Ingesting data...")
    import pandas as pd
    df = pd.read_csv(config["data_path"])
    assert len(df) > 0, "Empty dataset"
    logger.info(f"Loaded {len(df)} rows, {len(df.columns)} columns")
    return config["data_path"]

def step_clean(data_path: str) -> str:
    """Step 2: Clean data."""
    logger.info("Cleaning data...")
    import pandas as pd
    df = pd.read_csv(data_path)
    initial_rows = len(df)
    df = df.dropna(subset=[config["target_column"]])
    df = df.drop_duplicates()
    logger.info(f"Cleaned: {initial_rows} -> {len(df)} rows")
    clean_path = "data/cleaned.csv"
    df.to_csv(clean_path, index=False)
    return clean_path

def step_train(clean_path: str, config: dict) -> dict:
    """Step 3: Train model."""
    logger.info("Training model...")
    import pandas as pd
    import joblib
    from sklearn.model_selection import cross_val_score
    from sklearn.ensemble import GradientBoostingClassifier

    df = pd.read_csv(clean_path)
    X = df.drop(columns=[config["target_column"]])
    y = df[config["target_column"]]

    model = GradientBoostingClassifier(random_state=42)
    scores = cross_val_score(model, X, y, cv=5, scoring="roc_auc")
    logger.info(f"CV ROC AUC: {scores.mean():.4f} +/- {scores.std():.4f}")

    model.fit(X, y)
    model_path = "model_artifacts/model.joblib"
    Path("model_artifacts").mkdir(exist_ok=True)
    joblib.dump(model, model_path)

    return {"model_path": model_path, "cv_score": scores.mean()}

def step_evaluate(results: dict, config: dict) -> bool:
    """Step 4: Gate -- is the model good enough?"""
    logger.info("Evaluating model...")
    threshold = config.get("min_score", 0.8)
    passed = results["cv_score"] >= threshold
    if passed:
        logger.info(f"PASSED: {results['cv_score']:.4f} >= {threshold}")
    else:
        logger.warning(f"FAILED: {results['cv_score']:.4f} < {threshold}")
    return passed

def run_pipeline(config: dict):
    """Execute full pipeline."""
    logger.info(f"Pipeline started at {datetime.now().isoformat()}")
    try:
        data_path = step_ingest(config)
        clean_path = step_clean(data_path)

From tezansahu/ai-garage, GPL-3.0.

More from tezansahu/ai-garage