The Architecture Decision: Designing Your Pipeline Skeleton

Module 5, Unit 1 | Lesson 3 of 3

By the end of this lesson, you will be able to:

• Design a modular pipeline architecture with typed inputs and outputs at every stage boundary (K12, S10)
• Implement PipelineState and StageResult dataclasses that carry data and audit information through the full pipeline (K11, K12)
• Write five pipeline stage functions that chain together in a working runner with halt logic (S10, B6)
• Explain what makes a pipeline production-ready, and identify which properties your skeleton provides from day one (K11, S10)

You now have two building blocks: a data profile that documents what is wrong with the source data, and a Pydantic schema that enforces a clean contract at the input boundary. In this lesson you connect them inside a pipeline skeleton — a modular, typed structure that will carry all of your Unit 2, 3, and 4 logic.

A skeleton is not a placeholder. It is the architecture decision. The choices you make here — how stages communicate, what state they share, how errors propagate — will either make your pipeline easy to debug and extend, or make every future addition painful.

🔑 Key term — Pipeline skeleton: A working pipeline with the correct stage structure, typed interfaces, and wiring in place, but with placeholder logic in stages that will be filled in later. A skeleton lets you test the architecture before the logic exists, and lets different stages be developed independently without breaking the overall structure.

What makes a pipeline production-ready?

Before writing code, be explicit about what you are optimising for. A production pipeline needs four properties:

Modular — each stage does exactly one thing. You can test, replace, or debug a stage without touching anything else. If stage_validate changes, stage_classify does not need to know.

Typed — data flowing between stages has a defined shape. If Stage 2 changes what it produces, Stage 3 fails immediately with a clear error, rather than silently misreading a field three stages later.

Observable — every stage records what it did: how many records it processed, how long it took, and what happened to records that failed. Without this, debugging a production failure is guesswork.

Fail-fast — an error at any stage stops the pipeline cleanly, rather than propagating corrupt data downstream where the failure is harder to trace.

Coach Cora

These four properties are not aspirational. They are what separates a script you run once from a pipeline you trust in production. The Module 3 Commit Log Agent had none of them — and that was fine for a single-run prototype. Module 5 is about building the real thing, and these properties are non-negotiable from the start.

Design the stage contracts

Before writing the pipeline runner, define what flows between stages. Create contracts.py:

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any

from schema import ToolUsageRecord


@dataclass
class StageResult:
    """The audit record left behind by a single pipeline stage."""
    stage_name: str
    records_in: int
    records_out: int
    records_failed: int
    duration_seconds: float
    errors: list[dict[str, Any]] = field(default_factory=list)
    metadata: dict[str, Any] = field(default_factory=dict)

Every stage produces one of those, whatever it does. The second dataclass is what travels between them — one object that accumulates as the run proceeds, so no stage has to know which stage ran before it:

@dataclass
class PipelineState:
    """Accumulated state passed through the full pipeline."""
    run_id: str
    started_at: datetime
    source_file: str
    raw_rows: list[dict[str, Any]] = field(default_factory=list)
    valid_records: list[ToolUsageRecord] = field(default_factory=list)
    classified_records: list[dict[str, Any]] = field(default_factory=list)
    routed_records: list[dict[str, Any]] = field(default_factory=list)
    stage_results: list[StageResult] = field(default_factory=list)
    halted: bool = False
    halt_reason: str = ""

PipelineState is the envelope that travels through every stage. Each stage reads from it, does its work, and writes results back in. StageResult is the record each stage leaves behind — this is what powers your audit trail in Unit 3.

Click to enlarge

Curious Cat

Why not just pass a list of records from stage to stage? Because a list of records carries no context. When something breaks at 2am, you need to know: what file was being processed, when did each stage run, how many records were rejected and why. PipelineState carries all of that alongside the data itself — and because it travels through every stage, any stage can add to it without the others needing to change.

Write the stage functions

Create stages.py. Its import header first — every stage in the module lands in this one file, and this is the whole list it ever needs from the standard library and Pydantic:

from __future__ import annotations

import csv
import time
from datetime import datetime
from typing import Any

from pydantic import ValidationError

from contracts import PipelineState, StageResult
from schema import ToolUsageRecord

Then the first stage. Each function takes PipelineState, does one job, appends a StageResult, and returns the modified state:

def stage_ingest(state: PipelineState) -> PipelineState:
    """Stage 1: Read the source CSV into raw row dicts."""
    started = time.time()
    rows = []

    with open(state.source_file, newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for i, row in enumerate(reader, start=1):
            rows.append({"_source_row": i, **dict(row)})

    state.raw_rows = rows

    state.stage_results.append(StageResult(
        stage_name="ingest",
        records_in=0,
        records_out=len(rows),
        records_failed=0,
        duration_seconds=round(time.time() - started, 3),
        metadata={"source_file": state.source_file},
    ))

    print(f"[ingest]   {len(rows)} rows read from {state.source_file}")
    return state

Note the shape, because all five stages share it: take the state, do one job, append a StageResult, print a line for the human watching, return the state. records_in=0 for ingest is not a mistake — nothing flows into the first stage; it reads from the file.

The remaining four stages are collapsed below, one per stage. Open the one you are writing; the labels say what is inside each. Nothing is hidden that you need in order to follow the argument — and the whole file is in the download at the end of this lesson.

Stage 2 — validate. This is the only stage in the skeleton with real logic, because it is the only one whose job is already fully specified: run each row through the schema you wrote in L1.2.

def stage_validate(state: PipelineState) -> PipelineState:
    """Stage 2: Validate and normalise raw rows against the Pydantic schema."""
    started = time.time()
    valid = []
    errors = []

    for row in state.raw_rows:
        try:
            record = ToolUsageRecord(**{
                k: v for k, v in row.items()
                if not k.startswith("_")
            })
            valid.append(record)
        except ValidationError as e:
            errors.append({
                "log_id": row.get("log_id", "UNKNOWN"),
                "source_row": row.get("_source_row"),
                # The messages, not e.errors() itself. A Pydantic error entry
                # carries the original exception object under "ctx", which no
                # JSON encoder can write — and this list is bound for the
                # output file in L2.3.
                "errors": [err["msg"] for err in e.errors()],
            })

    state.valid_records = valid

    # Halt if error rate exceeds threshold
    error_rate = len(errors) / len(state.raw_rows) if state.raw_rows else 0
    if error_rate > 0.10:
        state.halted = True
        state.halt_reason = (
            f"Validation error rate {error_rate:.0%} exceeds 10% threshold "
            f"({len(errors)} of {len(state.raw_rows)} records failed)"
        )

    state.stage_results.append(StageResult(
        stage_name="validate",
        records_in=len(state.raw_rows),
        records_out=len(valid),
        records_failed=len(errors),
        duration_seconds=round(time.time() - started, 3),
        errors=errors,
    ))

    print(f"[validate] {len(valid)} valid, {len(errors)} failed")
    if state.halted:
        print(f"[validate] HALTING: {state.halt_reason}")

    return state

The halt threshold is a judgement call encoded as a number. Ten per cent says: a handful of bad rows is normal operational noise, but if a tenth of the file is wrong, something has changed upstream and continuing would produce a confident-looking result built on data you do not understand.

Stages 3 and 4 — the placeholders. These do nothing yet, and that is the point. They establish the shape so Units 2 and 3 can fill in bodies without anything around them moving. Read the docstrings: they are the specification you will implement later.

def stage_classify(state: PipelineState) -> PipelineState:
    """
    Stage 3: Classify records using an LLM agent.

    Placeholder — Unit 3 (L3.1) implements the real agentic classifier.
    Records with flagged_issue == 'none' will be passed through without an LLM call.
    Flagged records will be classified into: compliance_risk, operational_issue,
    cost_anomaly, or data_quality.
    """
    started = time.time()

    classified = [
        {
            "record": r,
            "classification": "pending",
            "confidence": None,
            "llm_called": False,
        }
        for r in state.valid_records
    ]
    state.classified_records = classified

    state.stage_results.append(StageResult(
        stage_name="classify",
        records_in=len(state.valid_records),
        records_out=len(classified),
        records_failed=0,
        duration_seconds=round(time.time() - started, 3),
        metadata={"mode": "placeholder — real agent added in Unit 3"},
    ))

    print(f"[classify] {len(classified)} records (placeholder - no LLM calls yet)")
    return state

Stage 4 is thinner still — it stamps every record with a single default route. L2.2 replaces the body with real routing rules.

def stage_route(state: PipelineState) -> PipelineState:
    """Stage 4: Route classified records to the correct handler."""
    started = time.time()

    routed = [{"route": "default", **item} for item in state.classified_records]
    state.routed_records = routed

    state.stage_results.append(StageResult(
        stage_name="route",
        records_in=len(state.classified_records),
        records_out=len(routed),
        records_failed=0,
        duration_seconds=round(time.time() - started, 3),
        metadata={"mode": "placeholder — routing logic added in Unit 2"},
    ))

    print(f"[route]    {len(routed)} records routed (placeholder)")
    return state

Stage 5 — output. Serialising is fiddlier than it looks: PipelineState holds date objects and Enum members, and json.dump refuses both. The serialise helper passed as default= handles them.

def stage_output(state: PipelineState) -> PipelineState:
    """Stage 5: Write pipeline results to disk."""
    import json
    from datetime import date

    started = time.time()

    def serialise(obj: Any) -> Any:
        if isinstance(obj, date):
            return obj.isoformat()
        if hasattr(obj, "value"):
            return obj.value
        raise TypeError(f"Not serialisable: {type(obj)}")

    state.stage_results.append(StageResult(
        stage_name="output",
        records_in=len(state.routed_records),
        records_out=len(state.routed_records),
        records_failed=0,
        duration_seconds=round(time.time() - started, 3),
    ))

    output = {
        "run_id": state.run_id,
        "started_at": state.started_at.isoformat(),
        "source_file": state.source_file,
        "halted": state.halted,
        "halt_reason": state.halt_reason,
        "stages": [
            {
                "stage": r.stage_name,
                "records_in": r.records_in,
                "records_out": r.records_out,
                "records_failed": r.records_failed,
                "duration_s": r.duration_seconds,
            }
            for r in state.stage_results
        ],
        "total_valid": len(state.valid_records),
    }

    with open("pipeline_output.json", "w") as f:
        json.dump(output, f, indent=2, default=serialise)

    print(f"[output]   pipeline_output.json written")
    return state

🔑 Key term — Fail-fast: A design principle where a system detects errors as early as possible and stops immediately rather than continuing with potentially corrupt state. In your pipeline, the 10% error threshold in stage_validate is a fail-fast mechanism — it halts cleanly rather than producing a result based on 90% of the expected data, which could be misleading or dangerous.

Write the pipeline runner

Create pipeline.py — the entry point that chains all stages together:

from __future__ import annotations

import uuid
from datetime import datetime, timezone

from contracts import PipelineState
from stages import (
    stage_ingest,
    stage_validate,
    stage_classify,
    stage_route,
    stage_output,
)

PROCESSING_STAGES = [
    stage_ingest,
    stage_validate,
    stage_classify,
    stage_route,
]

stage_output is imported but deliberately left out of PROCESSING_STAGES. The loop below halts on the first stage that sets state.halted, and output is called after the loop — so a halted run still writes its evidence. That changes in L2.3, once the destination is an external system.

def run_pipeline(source_file: str) -> PipelineState:
    state = PipelineState(
        run_id=str(uuid.uuid4()),
        started_at=datetime.now(timezone.utc),
        source_file=source_file,
    )

    print(f"\n{'='*50}")
    print(f"Pipeline run: {state.run_id}")
    print(f"Source:       {state.source_file}")
    print(f"{'='*50}\n")

    for stage_fn in PROCESSING_STAGES:
        state = stage_fn(state)
        if state.halted:
            print(f"\nPipeline halted after [{stage_fn.__name__}].")
            break

    state = stage_output(state)

    total_time = sum(r.duration_seconds for r in state.stage_results)
    print(f"\n{'='*50}")
    print(f"Completed in {total_time:.3f}s  |  Valid: {len(state.valid_records)}")
    print(f"{'='*50}\n")

    return state


if __name__ == "__main__":
    run_pipeline("ai_tool_usage_log.csv")

Run it

python pipeline.py

==================================================
Pipeline run: a3f2c1d4-...
Source:       ai_tool_usage_log.csv
==================================================

[ingest]   30 rows read from ai_tool_usage_log.csv
[validate] 30 valid, 0 failed
[classify] 30 records (placeholder - no LLM calls yet)
[route]    30 records routed (placeholder)
[output]   pipeline_output.json written

==================================================
Completed in 0.021s  |  Valid: 30
==================================================

Your pipeline runs end-to-end. Stages 3 and 4 are explicit, labelled placeholders — adding real logic in Units 2 and 3 means replacing the body of those functions, not restructuring anything around them.

Your file structure at the end of Unit 1

module6_pipeline/
  ai_tool_usage_log.csv     ← source data
  schema.py                 ← Pydantic model and all validators
  contracts.py              ← PipelineState and StageResult
  stages.py                 ← one function per pipeline stage
  pipeline.py               ← runner: chains stages, handles halt
  profile.py                ← profiling script from L1.1
  validate.py               ← standalone validation script from L1.2
  data_profile.json         ← artefact: findings from L1.1
  clean_records.json        ← artefact: normalised records from L1.2
  pipeline_output.json      ← artefact: run summary from L1.3

Every file has a single responsibility. Every artefact is named, dated and traceable. This is what an auditable pipeline looks like before it is even doing anything interesting.

Build activity

Get the full pipeline running with python pipeline.py. Then verify three things:

1. The console shows all five stages completing in order with the correct record counts.
2. pipeline_output.json contains the correct run_id, all five stage summaries (ingest, validate, classify, route, and output), and "halted": false.
3. The halt condition works: temporarily corrupt 4 rows in the CSV to introduce validation failures — enough to push the error rate above 10%. Confirm that the pipeline halts after stage_validate, still writes pipeline_output.json, and shows "halted": true.

Restore the CSV before moving to Unit 2.

Note on the halt test. The runner above uses break, not return, so stage_output still runs after a halt and you get an output file recording the failed run. That is the right trade while the only destination is a local file: a record of the failed run beats no record. It changes in L2.3, once output goes through a connector that can itself be unavailable — from there a halted run writes nothing at all. The halt test you just ran will not hold after that lesson.

Challenge Chase

Right now the pipeline runner calls stages as a hardcoded list. Refactor pipeline.py so that stages are registered in a dict keyed by name, and the execution order is read from a pipeline_config.json file. This is how real orchestration frameworks work: the code stays the same, but configuration determines what runs and in what order. What are the trade-offs compared to the hardcoded list?

One question to take to your own work, with nothing to build: if a pipeline you rely on halted at 2am, who gets told? If you cannot name the person, that is a finding worth raising — and it is portfolio evidence of exactly the initiative B6 describes.

The code from this lesson

This is the whole Unit 1 pipeline, ready to run, as the lesson leaves it. The code files are assembled from the code blocks above, so they match what you were asked to write; the dataset and a short README come with them.

Take it if you would rather read and run the code than type it out. Typing it is still the better way to learn it — this is here so a typo cannot cost you the lesson.

Checklist

• contracts.py defines both StageResult and PipelineState with correct field types
• stages.py contains all five stage functions, each appending a StageResult to state.stage_results
• pipeline.py chains all five stages and halts cleanly if state.halted is set
• Running python pipeline.py produces the expected console output and writes pipeline_output.json
• I have tested the halt condition by introducing enough validation errors to exceed the 10% threshold
• I can explain why PipelineState is preferable to passing records directly between stages
• I can name all four properties that make a pipeline production-ready and explain what breaks when each one is missing

Quick Check

KSB evidence focus

• K11 — Understands relevant data governance, data privacy and security issues. PipelineState and StageResult form the foundation of your audit trail. Every processing decision — how many records were validated, how many failed, why the pipeline halted — is recorded automatically. This is the technical implementation of data governance: not a policy, but a mechanism.

• K12 — Understands how to set up, interact with and generate APIs, databases and spreadsheets. The pipeline architecture you have built is the same pattern used by production data engineering tools like Apache Airflow and Prefect. Understanding typed stage contracts and state management transfers directly to those environments.

• S10 — Can carry out data processing tasks, including processing/cleaning, data transformations, and feature engineering. Your pipeline now takes raw, messy CSV data through ingestion, schema validation, and normalisation — and produces clean, typed, auditable output. That end-to-end data processing flow is the core deliverable of S10 at developer level.

• B6 — Shows curiosity and initiative. You built a halt mechanism before you needed it, documented the stage contracts before filling in the logic, and left clear placeholders with descriptive docstrings for Unit 3. That is the habit of a developer who thinks ahead — not just someone who makes today's tests pass.

Up next: Unit 2 fills in the pipeline logic. You will decide which stages need deterministic workflow code and which need an LLM agent, implement real routing with conditional branches, add retry logic for transient failures, and design the connector interface that lets your pipeline write to any output destination.