pip install -r requirements.txtRun the included classification pipeline:
python -m src.pipeline_runner --config pipelines/classification_pipeline.yaml --verboseThis will:
1. Generate a demo dataset (2000 rows with age, income, region, etc.)
2. Apply feature engineering (scaling, encoding, imputation)
3. Train a Random Forest classifier
4. Evaluate with accuracy, F1, and AUC threshold gates
5. Attempt a canary deployment (dry-run — no actual infrastructure)
6. Save all artifacts to ./artifacts/
After the run completes, check:
./artifacts//run_summary.json — full run metadata./artifacts//snapshot.json — reproducibility snapshot1. Copy pipelines/classification_pipeline.yaml to pipelines/my_pipeline.yaml
2. Update the feature_engineering stage to match your columns
3. Adjust the training hyperparameters
4. Set meaningful threshold_gates in the evaluation stage
5. Configure your deployment endpoint in the deployment stage
Create a function with this signature and register it:
from src.pipeline_runner import register_stage
def my_custom_stage(params: dict, context: dict) -> dict:
# Do your work here
result = {"my_output": "some_value"}
return result
register_stage("my_custom_stage", my_custom_stage)Then reference it in your pipeline YAML:
stages:
- name: my_custom_stage
depends_on: [feature_engineering]
params:
my_param: my_valueArchitectural patterns and best practices for building ML pipelines with this framework.
Train multiple models in parallel, pick the best one:
stages:
- name: feature_engineering
depends_on: []
- name: train_rf
depends_on: [feature_engineering]
params:
framework: sklearn
model_type: random_forest
- name: train_gb
depends_on: [feature_engineering]
params:
framework: sklearn
model_type: gradient_boosting
- name: model_selection
depends_on: [train_rf, train_gb]The DAG runner executes train_rf and train_gb in parallel (same wave), then model_selection runs after both complete.
Compare new model against the current production model:
stages:
- name: training
depends_on: [feature_engineering]
- name: evaluation
depends_on: [training]
params:
comparison_run_path: ./artifacts/production_baseline/eval_results.json
threshold_gates:
accuracy: 0.85The evaluation stage loads the baseline metrics and computes deltas, so you can see exactly how the new model compares.
Deploy to staging first, then production:
stages:
- name: deploy_staging
depends_on: [evaluation]
params:
strategy: blue_green
endpoint_url: https://staging.example.com/v1/models/churn
- name: deploy_production
depends_on: [deploy_staging]
params:
strategy: canary
canary_percentage: 5
endpoint_url: https://api.example.com/v1/models/churnLoad features from your feature store instead of raw data:
from src.pipeline_runner import register_stage
def load_from_feature_store(params, context):
# Your feature store client here
import pandas as pd
entity_df = pd.read_csv(params["entity_source"])
# features = feast_client.get_historical_features(...)
return {"dataframe": entity_df}
register_stage("load_features", load_from_feature_store)Use environment overrides for grid search in CI:
for lr in 0.01 0.05 0.1; do
for depth in 4 6 8; do
MLOPS_PIPELINE__STAGES__1__PARAMS__HYPERPARAMETERS__LEARNING_RATE=$lr \
MLOPS_PIPELINE__STAGES__1__PARAMS__HYPERPARAMETERS__MAX_DEPTH=$depth \
python -m src.pipeline_runner --config pipelines/regression_pipeline.yaml
done
doneEach run gets its own run_id and snapshot, so you can compare results after.
| Anti-Pattern | Why It's Bad | What to Do Instead |
|---|---|---|
| Hardcoded file paths | Breaks in CI, other machines | Use config params |
| No threshold gates | Bad models reach production | Always set minimum thresholds |
| Skipping reproducibility | Can't debug regressions | Always capture snapshots |
| One giant stage | Can't retry/parallelize | Break into focused stages |
| No artifact versioning | Overwrite previous results | Use run_id-based paths |