Reference library

ML engineering pipelines

Feature prep, batch inference, model-serving hooks, and production ML workflow glue.

4 matches

How to Compute a Confusion Matrix in Python

Compute a multi-class confusion matrix from true and predicted labels using pure Python dictionaries and nested lists, then format it for readable output.

confusion-matrix classification ml-metrics

Python

from collections import defaultdict

def compute_confusion_matrix(y_true, y_pred, labels):
    """Compute confusion matrix using Python dicts and nested lists."""
    label_index = {label: i for i, label in enumerate(labels)}
    matrix = [[0] * len(labels) for _ in range(len(labels))]
    
    for true, pred in zip(y…

14 0 Open

ML engineering pipelines medium

How to Mock Kedro Pipeline Nodes in Python

Create a modular Kedro pipeline with node functions, namespacing, and input/output mapping to mock pipeline execution locally.

kedro pipeline modular

Python

from kedro.pipeline import Pipeline, node
from kedro.pipeline.modular_pipeline import pipeline as modular_pipeline


def preprocess(data: list) -> list:
    """Clean data by removing None values."""
    return [item for item in data if item is not None]


def transform(data: list) -> list:
    """Add 1 to each numeric…

15 0 Open

ML engineering pipelines easy

How to Mock Shadow Mode Inference in Python

Simulates running multiple candidate models in shadow mode by adding randomized delays and returning their outputs alongside a primary model's output.

ml-pipeline shadow-mode simulation

Python

import random
import time


def shadow_mode_inference(candidates, mock_delay=0.1):
    """
    Simulates running multiple candidate models in 'shadow mode'
    by adding tiny randomized delays and returning their outputs
    alongside the primary model's output.
    """
    primary_output = "primary: answer"
    shado…

13 0 Open

ML engineering pipelines medium

Mock a Flyte ML workflow in Python

Build a lightweight mock of a Flyte ML pipeline with dataclasses and a simple execution loop that passes outputs between tasks.

flyte ml-pipeline dataclass

Python

from dataclasses import dataclass, field
from typing import List, Dict, Optional
import time


@dataclass
class FlyteTask:
    name: str
    inputs: Dict = field(default_factory=dict)
    outputs: Dict = field(default_factory=dict)

    def run(self) -> Dict:
        time.sleep(0.1)  # simulate work
        return sel…

16 0 Open

Browse by section

Each section groups closely related Python snippets.