mle-workflow

ML Engineering Workflow

Turns model work into production ML systems. Use only the stages that match your system — don't force heavyweight MLOps onto a simple classifier.

When to Use

• Building a production ML feature (classifier, ranker, embeddings, LLM pipeline)
• Converting notebook code into a reproducible training pipeline
• Designing evaluation criteria before training starts
• Debugging data drift, stale features, or training/serving skew
• Planning model deployment and rollback

Stage 1 — Data Contract

Define before writing code. Everything downstream depends on this.

# data_contract.py
from dataclasses import dataclass
from typing import Optional

@dataclass
class TrainingExample:
    text: str                    # input feature
    label: str                   # target: "positive" | "negative" | "neutral"
    source: str                  # where this came from
    created_at: str              # ISO 8601 timestamp
    metadata: Optional[dict] = None

# Constraints (validate in data loader, not model)
LABEL_SET = {"positive", "negative", "neutral"}
MAX_TEXT_TOKENS = 512
MIN_EXAMPLES_PER_LABEL = 100

Document: feature schema, label definitions (with examples of edge cases), known data quality issues, train/val/test split strategy, and class balance.

Gate: do not proceed until you have ≥ MIN_EXAMPLES_PER_LABEL for every label.

Stage 2 — Baseline First

Before trying a complex model, establish a baseline you can beat:

• Rule-based: keyword matching, regex, heuristics
• Classical ML: TF-IDF + logistic regression, or fastText
• Pretrained zero-shot: test the LLM without fine-tuning first

Baselines are not just warm-ups. They define the performance floor and often expose that a simpler model is good enough.

from sklearn.linear_model import LogisticRegression
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.pipeline import Pipeline
from sklearn.metrics import classification_report

baseline = Pipeline([
    ("tfidf", TfidfVectorizer(max_features=10_000)),
    ("clf", LogisticRegression(max_iter=1000))
])
baseline.fit(X_train, y_train)
print(classification_report(y_test, baseline.predict(X_test)))