AiCore logo

Assembling the Complete Pipeline

Module 5, Unit 4 | Lesson 1 of 3

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

  • Wire all six stages and supporting modules into a single runnable pipeline (K11, K12, S10, S11)
  • Verify the complete pipeline runs end-to-end on the 30-record dataset without errors (S10, S11)
  • Read and interpret the full pipeline_output.json to confirm correct behaviour at every stage (K11, S10)
  • Identify and fix any integration issues that only appear when all stages run together (K12, S11, B6)

Across Units 1, 2, and 3 you built each piece of the pipeline separately: the schema, the stages, the connectors, the classifiers, the logger. In this lesson you wire all of it together and run the complete system end-to-end for the first time.

This is integration work — and integration always reveals things that isolated unit tests do not.

🔑 Key term — Integration: Running multiple components together as a connected system. Individual unit tests verify that each component works in isolation. Integration testing verifies that they work when connected — that the output of stage_validate is the correct input format for stage_classify, that the logger initialised in pipeline.py is available inside stage_route, and so on.

Your final file structure

Before wiring, confirm your project contains all these files:

module6_pipeline/
├─ .venv/                          # Virtual environment (L1.1, git-ignored)
├─ .env                            # OPENAI_API_KEY (L3.1, git-ignored)
├─ .gitignore                      # L1.1
├─ requirements.txt                # L1.1
├─ ai_tool_usage_log.csv           # Source dataset
├─ schema.py                       # Pydantic ToolUsageRecord model
├─ contracts.py                    # PipelineState and StageResult dataclasses
├─ stages.py                       # Stage functions + utilities (stage_output moves to pipeline.py)
├─ classification_contract.py      # Shared labels, prompt and threshold (L3.3)
├─ classifier.py                   # OpenAI cloud classifier
├─ local_classifier.py             # Ollama local classifier
├─ connectors.py                   # OutputConnector ABC + JSONFileConnector
├─ pipeline_logger.py              # PipelineLogger (JSONL writer)
├─ pipeline.py                     # Runner: orchestrates all stages
├─ test_workflow.py                # Unit tests (L2.1)
├─ profile.py                      # Profiling script (L1.1)
├─ validate.py                     # Standalone validation script (L1.2)
├─ data_profile.json               # Artefact: L1.1 findings
├─ clean_records.json              # Artefact: L1.2 normalised records
├─ logs/                           # Generated log files
└─ pipeline_output.json            # Generated output (created on first run)

If any file is missing, refer back to the lesson that created it before continuing.

Note on stage_output: The simple stage_output you wrote in L1.3's stages.py is replaced in this lesson by the full version defined directly in pipeline.py. Before running, delete stage_output from stages.py to avoid a naming conflict.

The complete pipeline.py

Here is the full, wired pipeline.py — consolidating everything:

from __future__ import annotations

from dotenv import load_dotenv
load_dotenv()  # Must be before any import that creates the OpenAI client at module level

import time
import uuid
from datetime import datetime, timezone

from connectors import JSONFileConnector
from contracts import PipelineState, StageResult
from pipeline_logger import PipelineLogger
from stages import (
    output_records,
    run_metadata,
    stage_ingest,
    stage_validate,
    stage_classify,
    stage_route,
    stage_human_approval,
)

# Output is handled separately after the loop, so the runner can inject the connector.
PROCESSING_STAGES = [
    stage_ingest,
    stage_validate,
    stage_classify,
    stage_route,
    stage_human_approval,
]

Read the import block before you paste it. load_dotenv() sits above every other import, which looks wrong and is deliberate: it guarantees the environment is populated before anything else runs. Since L3.1 the OpenAI client is built lazily by get_client(), so this is belt-and-braces rather than load-bearing — but keep it there. The moment any dependency does read configuration at import time, this line is the difference between a working pipeline and an authentication error pointing at the wrong file.

PROCESSING_STAGES holds five stages, not six. Output is deliberately outside the list so the runner can hand it a connector — the same split you set up in L1.3.

Stage 6 — output. Two things change here. The payload gains the classifier's own fields and becomes fully JSON-serialisable, and the stage itself moves out of stages.py into pipeline.py, because it is the only stage that needs a connector.

Update output_records in stages.py — the version from L2.3 passed date and Enum objects straight through and left the connector to serialise them. Now that the payload is what an auditor reads, do the conversion here where the field names are visible:

def output_records(state: PipelineState) -> list[dict[str, Any]]:
    """Flatten routed records into the shape a connector sends."""
    return [
        {
            "log_id": item["record"].log_id,
            "date": item["record"].date.isoformat(),
            "tool_name": item["record"].tool_name.value,
            "flagged_issue": item["record"].flagged_issue.value,
            "route": item.get("route", "unknown"),
            "classification": item.get("classification", "pending"),
            "confidence": item.get("confidence"),
            "model_used": item.get("model_used", "none"),
            "cost_usd": float(item["record"].cost_usd),
            "human_reviewed": item["record"].human_reviewed,
        }
        for item in state.routed_records
    ]

confidence and model_used come from L3.1 and L3.3. A record that never reached a model carries "model_used": "none" rather than a missing key, so a reader can tell "not classified" from "classified by something we forgot to record".

Now stage_output itself, in pipeline.py. Structurally it is the L2.3 version with logging added:

def stage_output(state: PipelineState, connector=None) -> PipelineState:
    """Stage 6: Write pipeline results via the configured connector."""
    started = time.time()

    if connector is None:
        connector = JSONFileConnector("pipeline_output.json")

    if not connector.health_check():
        state.halted = True
        state.halt_reason = (
            f"Output connector health check failed: "
            f"{connector.__class__.__name__}"
        )
        state.logger.error(
            "output", "connector_unavailable",
            connector=connector.__class__.__name__,
        )
        print("[output]   HALTING - connector unavailable")
        return state

    records = output_records(state)

    state.stage_results.append(StageResult(
        stage_name="output",
        records_in=len(state.routed_records),
        records_out=len(records),
        records_failed=0,
        duration_seconds=round(time.time() - started, 3),
        metadata={"connector": connector.__class__.__name__},
    ))

    connector.write(records, run_metadata(state, len(records)))
    state.logger.info(
        "output", "pipeline_complete",
        records_written=len(records),
        connector=connector.__class__.__name__,
    )

    return state

Two details that matter for your audit trail. The StageResult is appended before connector.write() is called, which is why the output stage appears in its own stages array — append it afterwards and the file would describe a five-stage run. And the health check halts before writing anything, so a partial write to an unreachable destination is impossible.

The runner. Everything above is wiring; this is the part that decides what happens when something goes wrong:

def run_pipeline(source_file: str, connector=None) -> PipelineState:
    """Run the complete six-stage pipeline."""
    run_id = str(uuid.uuid4())[:8]

    state = PipelineState(
        run_id=run_id,
        started_at=datetime.now(timezone.utc),
        source_file=source_file,
        logger=PipelineLogger(
            log_path=f"logs/pipeline_{run_id}.jsonl",
            run_id=run_id,
        ),
    )

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

    for stage_fn in PROCESSING_STAGES:
        state = stage_fn(state)
        if state.halted:
            state.logger.error(
                "pipeline", "halted",
                reason=state.halt_reason,
            )
            print(f"\n[pipeline] HALTED: {state.halt_reason}")
            return state

    # Output sits outside PROCESSING_STAGES so the runner can inject the connector.
    if connector is None:
        connector = JSONFileConnector("pipeline_output.json")
    state = stage_output(state, connector=connector)

    if not state.halted:
        state.logger.info("pipeline", "run_complete", run_id=run_id)
        print(f"\n{'='*50}")
        print(f"Pipeline complete - run_id: {run_id}")
        print(f"{'='*50}\n")

    return state


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

Read the two halt paths carefully, because they are not the same path. The stage loop returns on a halt rather than breaking, so a halted stage never reaches the write — that is how the L2.3 rule is honoured here: a halted run delivers nothing to an external destination. The if not state.halted further down covers the other case, where stage_output itself halts on a failed health check and the run must not announce that it completed.

Coach Cora

Coach Cora

The if name == "main" guard means run_pipeline only executes when you run pipeline.py directly — not when another module imports it. This is a Python convention, but it matters here: if a future script imports run_pipeline to test it, the import will not trigger a real pipeline run. Always include this guard in any script that has a meaningful side effect at module level.

Run the complete pipeline

cd module6_pipeline
python pipeline.py

Expected output sequence:

==================================================
Pipeline run: a1b2c3d4
Source: ai_tool_usage_log.csv
==================================================
[ingest]   30 rows read, 0 empty rows skipped
[validate] 30 valid, 0 invalid
[classify] 30 classified (4 cloud, 2 local, 24 skipped clean, 0 errors)
[route]    0 human_review | 2 compliance | 4 ops_review | 24 standard

==================================================
HUMAN APPROVAL REQUIRED
2 record(s) requiring approval:
==================================================
  LOG-006 | L017 | GPT-4 | 2024-01-09
  LOG-023 | L088 | Gemini Pro | 2024-01-14
==================================================

Approve processing these records? [yes/no]: yes

[approval] Approved. Continuing pipeline.
[connector] 30 records -> pipeline_output.json

==================================================
Pipeline complete - run_id: a1b2c3d4
==================================================

Your exact numbers will vary depending on how the LLM classifies each record.

Verify the output

Open pipeline_output.json and check:

  1. Total records matches records_out from the validate stage
  2. Every record has a route, classification, confidence, and model_used field
  3. PII-flagged records show "model_used": "ollama/qwen3:1.7b"
  4. Clean records show "classification": "none" and "confidence": 1.0
  5. The stages array lists all six stages (ingest through output) with durations, record counts, and per-stage metadata

If any field is missing or null unexpectedly, use the structured log to trace the affected records back through the pipeline.

Curious Cat

Curious Cat

Why does the classify stage run before the human approval stage? Because you need the classification to know which records require approval — you cannot ask "do you approve these records?" until the pipeline has identified compliance cases and low-confidence cases. The ordering of your stages encodes a dependency: route depends on classify, approval depends on route, output depends on approval. If the order changes, the logic breaks.

Common integration issues

If the pipeline halts or produces unexpected output, check these first:

  • ModuleNotFoundError for dotenv (or any other package): your virtual environment is not active. Check for (.venv) in your prompt; if it is missing, re-activate it (source .venv/bin/activate) and run pip install -r requirements.txt. Do not reach for --break-system-packages — that installs into your system Python and papers over the real problem.
  • ConnectionRefusedError from local_classifier: Ollama is not running. Run ollama serve in a separate terminal
  • OPENAI_API_KEY not found: Your .env file is missing, is in the wrong directory, or the load_dotenv() call is not at the top of pipeline.py. Note that this now surfaces on the first classification, not at import — get_client() is lazy
  • All records classified as "pending": The stage_classify stub from L2.1 is still in stages.py. Replace it with the real implementation from L3.1, as superseded by the dual-model version in L3.3
  • logs/ directory does not exist: The logger creates it automatically — if you see a permission error, create the folder manually

Build activity

  1. Confirm your file structure matches the list above. Create any missing files before continuing.
  2. Run python pipeline.py end-to-end. Do not stop until you see "Pipeline complete" in the terminal.
  3. Open pipeline_output.json and verify the required output fields across all 30 records and the route distribution.
  4. Open the JSONL log and trace one PII-flagged record from ingest through to output.
  5. Re-run the pipeline and enter no at the approval checkpoint. The run halts before stage_output, so no new pipeline_output.json is written — the file on disk is still the one from your last successful run. Do not read it and conclude the halt failed. Instead, open the new logs/pipeline_{run_id}.jsonl for this run and confirm it ends with a "halted" entry at level ERROR carrying the rejection reason. That log is the only evidence a rejected run leaves behind, which is exactly why L3.4 came before this lesson.
Challenge Chase

Challenge Chase

Your pipeline currently runs all stages sequentially, even stages that could be parallelised. Analyse the dependency graph: which stages must run after which? Draw it out. Then identify which stages (if any) could run concurrently — for example, could a second batch of records start ingestion while the first batch is in the classify stage? What data structures would need to change to support a streaming model rather than a batch model?

The code from this lesson

The complete six-stage pipeline, ready to run. The code files are assembled from the code blocks of this lesson and every lesson before it, so they match what you were asked to write.

Download the L4.1 code

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

  • All files in the project structure are present and correct
  • python pipeline.py runs end-to-end without errors or unhandled exceptions
  • The terminal output shows all six stages completing in the correct order
  • pipeline_output.json contains all expected fields for each output record
  • PII-flagged records show model_used: "ollama/qwen3:1.7b" in the output
  • The JSONL log contains entries from all six stages for the run
  • Re-running with no at the approval prompt halts the pipeline, writes no new pipeline_output.json, and records the halt in that run's JSONL log

After wiring all stages together, your pipeline runs but the output file contains only 20 records from 30 ingested. Every stage in pipeline_output.json — including output itself — shows records_in=30 and records_out=30. Where do you look?

KSB evidence focus

  • K12 — Understands how to set up, interact with and generate APIs, databases and spreadsheets. Running the complete pipeline for the first time integrates three external systems: the OpenAI API, the local Ollama API, and the JSON file output connector. The ability to wire these together in a single runner, handle their failure modes, and produce consistent output from all three is the integration skill this module is designed to build.

  • S10 — Can carry out data processing tasks, including processing/cleaning, data transformations, and feature engineering. The output verification step — checking field completeness, route distribution, and model attribution across all 30 records — is a data quality check applied to your own pipeline's output. You are treating the pipeline output as data, validating it the same way you validated the source CSV in Unit 1.

  • S11 — Can design, implement and test an AI/ML model or system. A pipeline that runs end-to-end without errors, produces complete and correctly structured output, and behaves predictably when halted is a working AI system. This is the milestone that moves the project from development to verification.

Up next: The pipeline works — now make it fast and cost-efficient. Lesson 2 measures the actual performance of each stage, identifies the bottlenecks, and applies batching and concurrency improvements where they matter most.

Next: L4.2 - Performance, Batching and Optimisation