The Universal Workflow of Machine Learning

From Problem Definition to Production MLOps

Success in AI isn't just about the model—it's about the **process**. Whether you're building a diagnostic tool or a perception stack, you need a repeatable workflow to manage data, scale training, and monitor production performance.

In this chapter, we explore the modern **MLOps** lifecycle, focusing on Data Flywheels, Concept Drift, and the ethics of deployment in safety-critical systems.

Step 1: Defining the Task

The first step is framing the problem. What are the inputs and outputs? What is the cost of a mistake? In an Autonomous Vehicle, a **False Negative** (missing a pedestrian) is much more costly than a **False Positive** (braking for a shadow). Choosing the right metric—like **Recall**—is part of the framing.

PythonRuns entirely in your browser — nothing is sent to a server.

# Simulation: Framing Decisions (Medical vs. AV)
class WorkflowArchitect:
    def __init__(self, scenario_type):
        self.scenario = scenario_type

    def recommend_strategy(self):
        if self.scenario == "Diagnostics":
            return {
                "Loss Function": "Binary Crossentropy",
                "Primary Metric": "Recall (Minimize False Negatives)",
                "Constraint": "High confidence required for surgical flagging"
            }
        elif self.scenario == "AV_Braking":
            return {
                "Loss Function": "Mean Squared Error (Distance)",
                "Primary Metric": "Latency (Inference < 10ms)",
                "Constraint": "Must handle edge cases like fog/rain"
            }
        return "Generic Strategy"

# Scenario: Clinical Pathology
arch = WorkflowArchitect("Diagnostics")
print(f"Strategy for {arch.scenario}:")
for k, v in arch.recommend_strategy().items():
    print(f" - {k}: {v}")

# Simulation: Framing Decisions (Medical vs. AV)
class WorkflowArchitect:
    def __init__(self, scenario_type):
        self.scenario = scenario_type

    def recommend_strategy(self):
        if self.scenario == "Diagnostics":
            return {
                "Loss Function": "Binary Crossentropy",
                "Primary Metric": "Recall (Minimize False Negatives)",
                "Constraint": "High confidence required for surgical flagging"
            }
        elif self.scenario == "AV_Braking":
            return {
                "Loss Function": "Mean Squared Error (Distance)",
                "Primary Metric": "Latency (Inference < 10ms)",
                "Constraint": "Must handle edge cases like fog/rain"
            }
        return "Generic Strategy"

# Scenario: Clinical Pathology
arch = WorkflowArchitect("Diagnostics")
print(f"Strategy for {arch.scenario}:")
for k, v in arch.recommend_strategy().items():
    print(f" - {k}: {v}")

Step 2: Developing the Model

Development starts with a **Baseline**—the simplest possible model that beats a common-sense hurdle (like guessing the most frequent class). From there, we scale up to a model with "Capacity" to solve the task, and then we regularize. In Medical AI, a baseline might be "always predict the patient has no disease"—your model *must* beat this significantly to be useful.

PythonRuns entirely in your browser — nothing is sent to a server.

# Simulation: Beating the Baseline
def get_baseline_accuracy(data_labels):
    # The 'Frequency' baseline
    most_common = max(set(data_labels), key=data_labels.count)
    correct = data_labels.count(most_common)
    return (correct / len(data_labels)) * 100

# Data: 90 Benign, 10 Malignant
medical_data = ["Benign"] * 90 + ["Malignant"] * 10
baseline = get_baseline_accuracy(medical_data)

print(f"Common-sense Baseline: {baseline}%")
print("Model Metric: To be 'useful', your model must achieve > 95% Balanced Accuracy.")

# Simulation: Beating the Baseline
def get_baseline_accuracy(data_labels):
    # The 'Frequency' baseline
    most_common = max(set(data_labels), key=data_labels.count)
    correct = data_labels.count(most_common)
    return (correct / len(data_labels)) * 100

# Data: 90 Benign, 10 Malignant
medical_data = ["Benign"] * 90 + ["Malignant"] * 10
baseline = get_baseline_accuracy(medical_data)

print(f"Common-sense Baseline: {baseline}%")
print("Model Metric: To be 'useful', your model must achieve > 95% Balanced Accuracy.")

Step 3: Deploying and Monitoring

Deployment isn't the end. Models in the wild encounter **Concept Drift**—the world changes, but the model doesn't. An AV vision system trained in sunny California may fail in a snowy German winter. We must monitor confidence scores and data distributions to know when to retrain.

PythonRuns entirely in your browser — nothing is sent to a server.

# Simulation: Model Health Monitor (Detecting Drift)
class ModelHealthMonitor:
    def __init__(self, confidence_threshold=0.85):
        self.threshold = confidence_threshold
        self.recent_scores = []

    def record_inference(self, score):
        self.recent_scores.append(score)
        if len(self.recent_scores) > 10: self.recent_scores.pop(0)
        
        avg_confidence = sum(self.recent_scores) / len(self.recent_scores)
        if avg_confidence < self.threshold:
            return f"ALERT: Avg Confidence {avg_confidence:.2f} < Threshold. Detecting Concept Drift! (Winter/Night?)"
        return f"STATUS: Healthy (Avg Conf: {avg_confidence:.2f})"

monitor = ModelHealthMonitor()
# Simulated inferences during a weather change (Confidence drops)
scores = [0.95, 0.92, 0.94, 0.88, 0.75, 0.72, 0.68, 0.65]

for s in scores:
    print(monitor.record_inference(s))

# Simulation: Model Health Monitor (Detecting Drift)
class ModelHealthMonitor:
    def __init__(self, confidence_threshold=0.85):
        self.threshold = confidence_threshold
        self.recent_scores = []

    def record_inference(self, score):
        self.recent_scores.append(score)
        if len(self.recent_scores) > 10: self.recent_scores.pop(0)
        
        avg_confidence = sum(self.recent_scores) / len(self.recent_scores)
        if avg_confidence < self.threshold:
            return f"ALERT: Avg Confidence {avg_confidence:.2f} < Threshold. Detecting Concept Drift! (Winter/Night?)"
        return f"STATUS: Healthy (Avg Conf: {avg_confidence:.2f})"

monitor = ModelHealthMonitor()
# Simulated inferences during a weather change (Confidence drops)
scores = [0.95, 0.92, 0.94, 0.88, 0.75, 0.72, 0.68, 0.65]

for s in scores:
    print(monitor.record_inference(s))

Practice Questions

Question 1

In the AV Braking scenario, why might 'Recall' be prioritized over 'Precision'?

Precision is too hard to calculate for cars
It is safer to occasionally brake for a 'shadow' (False Positive) than to miss a 'pedestrian' (False Negative)
Recall makes the engine run faster
AVs only use Recall

Question 2

What is 'Concept Drift' in the context of model deployment?

When the model file gets too large to fit on a server
When the relationship between input data and output targets changes over time in the real world
When the model starts taking too long to predict
When the code gets deleted by accident