When It Breaks: Systematic Pipeline Debugging

Module 5, Unit 3 | Lesson 5 of 5

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

• Apply a systematic four-step debugging framework to any pipeline failure (K11, S10, S11)
• Use your structured logs to locate the stage, record, and cause of a failure without re-running the full pipeline (S10, S11)
• Write targeted isolation tests that reproduce a specific failure at the stage level rather than the full pipeline level (K12, S10, S11)
• Distinguish between the four classes of AI pipeline failure and know where to look for each (K11, K12, S11)

A pipeline that has never failed in production is a pipeline that has never run long enough. Validation errors, API rate limits, model hallucinations, connector outages, schema mismatches — these are not hypothetical edge cases. They are normal operating conditions. The question is not whether your pipeline will fail, but how quickly you can understand and fix it when it does.

This lesson teaches you a systematic approach to debugging, using the structured logs and stage isolation capabilities you have built.

🔑 Key term — Root cause analysis: The process of identifying the underlying reason for a failure rather than fixing its symptoms. You already have the toolkit for this: Module 2, Unit 1, Lesson 1 taught the 5 Whys for drilling down a single causal chain and the Fishbone (Ishikawa) diagram for mapping contributing factors across categories. Nothing about those methods changes because the failure is in code — the object of study is a stack trace rather than a business process, and that is the only difference.

The worked example runs the same way. A record with an unexpected date format fails validation; re-run it and it fails again, so nothing is flaky. But stop at "the date was malformed" and you have a symptom. Keep going and you reach a source system that changed its export format, which will send you malformed dates every night from now on. That is precisely the failure mode Module 2 warns about — stopping at the first plausible answer and shipping a fix to a symptom dressed up as a fix to a cause.

The four classes of pipeline failure

Every failure in a data pipeline falls into one of four categories. Knowing the category tells you where to look.

1. Input failure — The source data violates your schema expectations. New date formats, unexpected null fields, character encoding issues, truncated rows. These always appear in the validate stage.

2. Infrastructure failure — A service your pipeline depends on is unavailable or behaves unexpectedly. An API is down, Ollama is not running, the output directory does not exist, a webhook returns 5xx. These appear as exceptions in the stage that makes the external call.

3. Model failure — The LLM returns an unexpected response: malformed JSON, an out-of-vocabulary classification label, a confidence score that is a string instead of a float, or reasoning that is an empty string. These appear in stage_classify after the API call succeeds.

4. Logic failure — Your pipeline code has a bug: a routing rule that catches the wrong records, a batch calculation that divides by zero for empty queues, a connector that silently drops records. These produce no exception — the pipeline runs to completion but produces wrong output.

Coach Cora

Logic failures are the hardest to find because the pipeline does not tell you something went wrong — you have to notice the output is incorrect. In a deliberately broken run, if your pipeline ingested 30 records, validated 27, classified 27, routed 27, and output 24, where did the three records go? That gap between records_in and records_out in each StageResult is your first signal.

The four-step debugging framework

When your pipeline fails or produces unexpected output, work through these steps in order. Do not skip ahead.

Step 1: Read the stage summary

Check the stage results in pipeline_output.json. Each stage shows records_in, records_out, and records_failed. The stage where records_failed > 0 or where records_out != records_in (without explanation) is where the problem lives.

# debug_run.py
import json

with open("pipeline_output.json") as f:
    output = json.load(f)

for stage in output["metadata"]["stages"]:
    # ingest has no upstream stage, so its records_in is 0 by design and a gap
    # here would mean nothing. Every other stage should read gap=0.
    gap = "n/a" if stage["stage"] == "ingest" else (
        stage["records_in"] - stage["records_out"]
    )
    print(
        f"{stage['stage']:20s} in={stage['records_in']} "
        f"out={stage['records_out']} failed={stage.get('records_failed', 0)} "
        f"gap={gap}"
    )

Step 2: Find the failing records in the structured log

Once you know which stage failed, filter the JSONL log for level=WARNING or level=ERROR entries from that stage:

import glob
import json
import os

# The newest pipeline run, which is nearly always the one you just broke.
# pipeline_*.jsonl rather than *.jsonl, so an isolation run from L3.5 or a
# benchmark run from L4.2 cannot be mistaken for it. Point this at a specific
# file when you are chasing an older run.
runs = sorted(glob.glob("logs/pipeline_*.jsonl"), key=os.path.getmtime)
if not runs:
    raise SystemExit("No pipeline logs in logs/ yet - run python pipeline.py first.")
LOG_FILE = runs[-1]

with open(LOG_FILE, encoding="utf-8") as f:
    entries = [json.loads(line) for line in f if line.strip()]

# Find all errors in the validate stage
errors = [
    e for e in entries
    if e["stage"] == "validate" and e["level"] in ("WARNING", "ERROR")
]
for e in errors:
    print(f"  {e['log_id']}: {e['data'].get('errors')}")

Step 3: Reproduce the failure in isolation

Once you have identified the failing record and stage, reproduce the failure without running the full pipeline. This is why each stage function accepts PipelineState — you can call a single stage with a minimal state:

# isolate_validate.py
from contracts import PipelineState
from stages import stage_validate
from datetime import datetime, timezone
from pipeline_logger import PipelineLogger
import uuid

# Construct a minimal state with just the failing row
run_id = "debug_" + str(uuid.uuid4())[:6]
state = PipelineState(
    run_id=run_id,
    started_at=datetime.now(timezone.utc),
    source_file="debug",
    logger=PipelineLogger(f"logs/{run_id}.jsonl", run_id),
)
state.raw_rows = [
    {
        "_source_row": 7,
        "log_id": "LOG-007",
        "date": "January 8 2024",      # ← the deliberate fault: unrecognised date format
        "learner_id": "L002",
        "tool_name": "GPT-4",
        "task_type": "summarisation",
        "model_used": "gpt-4",
        "prompt_tokens": "500",
        "completion_tokens": "200",
        "response_time_ms": "2100",
        "quality_score": "4.0",
        "time_saved_mins": "20",
        "flagged_issue": "none",
        "human_reviewed": "yes",
        "cost_usd": "0.048",
    }
]

result = stage_validate(state)
print(f"Valid: {len(result.valid_records)}")
print(f"Invalid: {len(result.invalid_rows)}")
for row in result.invalid_rows:
    print(f"  Errors: {row['errors']}")

Running this in isolation lets you fix and re-run the single failing case without touching the full dataset.

Every other field in that row is deliberately valid — "GPT-4" is in TOOL_NAME_MAP, "4.0" is inside the 1.0–5.0 range, "yes" parses to a boolean. Only the date is wrong. An isolation test with three faults in it tells you three things at once and teaches you nothing; change one variable so the error message you get can only mean one thing. If you had left tool_name as "ChatGPT" you would be debugging two failures and guessing which one you actually came here for.

Step 4: Fix the root cause, not the symptom

Before choosing a fix, run the 5 Whys on what you have reproduced. On the seeded date failure it goes something like this:

1. Why did the record vanish from the output? Validation rejected it, so it never reached the classify stage. (Corrupt four rows rather than one and the same chain starts why did the run halt? — four of thirty crosses the 10% threshold, one of thirty does not.)
2. Why were they rejected? The date field did not match any format the validator parses.
3. Why not? It arrived as January 8 2024 — a written-out month, which is not one of the three patterns in schema.py.
4. Why is a fourth format arriving at all? The export that produced this file changed, or a second source is now feeding the same pipeline.
5. Why did nobody know? Nothing in the pipeline reports the shape of incoming data, only whether it passed.

Notice that each answer suggests a different fix, and they get progressively less like patching and more like engineering. Stopping at (3) gets you a fourth date pattern and the same failure next quarter in a fifth format. Answering (5) gets you a validator that reports what it rejected and why — which is what the structured logging in L3.4 was for.

The four failure classes at the top of this lesson double as ready-made Fishbone categories when a fault has more than one contributing factor: instead of People / Process / Technology, you map candidate causes under Input, Infrastructure, Model and Logic. Once the cause is identified, the fix depends on which class it fell into:

• Input failure: update the Pydantic validator to handle the new format, or add a pre-processing step that normalises it before validation.
• Infrastructure failure: fix the configuration (wrong path, missing API key, Ollama not running), or add a graceful degradation path.
• Model failure: update the response parser to handle the unexpected format, add a stricter prompt instruction, or improve the validation logic.
• Logic failure: write a unit test that reproduces the wrong output, then fix the logic and verify the test passes.

Curious Cat

Why is "fix the root cause" a separate step? Because the most tempting response to a pipeline failure is to patch the symptom: add an except that swallows the error, or skip the failing record and move on. Both of these approaches hide the problem. A validator that silently drops records with unexpected date formats will continue dropping records from the new data source indefinitely. The gap in your output will grow. Fixing the root cause means the same failure does not recur.

Debugging a model failure specifically

Model failures deserve their own approach because the API call succeeded — the problem is in the response content, not the request. Add this diagnostic function to your toolkit:

# debug_classifier.py
from dotenv import load_dotenv
load_dotenv()

from classifier import classify_record

# Simulate what the model returned for a problematic record
# by running it again and inspecting the raw response

import os, json
from openai import OpenAI
from classifier import SYSTEM_PROMPT

client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))

record_data = {
    "log_id": "LOG-011",          # A real cost_overrun row from your CSV
    "tool_name": "GPT-4",
    "task_type": "data analysis",
    "flagged_issue": "cost_overrun",
    "cost_usd": 0.134,
    "time_saved_mins": 28,
    "quality_score": 3.6,
    "human_reviewed": True,
}

user_message = "\n".join(f"{k}: {v}" for k, v in record_data.items())

The call deliberately omits response_format. The production classifier constrains the model to a schema; here you want to see whatever it actually produces, including the malformed output the schema would have hidden.

response = client.chat.completions.create(
    model="gpt-4.1-mini",
    messages=[
        {"role": "system", "content": SYSTEM_PROMPT},
        {"role": "user", "content": f"Classify this AI usage log record:\n\n{user_message}"},
    ],
    temperature=0.1,
    max_completion_tokens=200,
)

Reading the reply is the point of the script. Print the raw string with repr before parsing it, and print finish_reason alongside — a truncated reply and a refusal look identical until you do.

raw = response.choices[0].message.content

print("Raw model response:")
print(repr(raw))
print("Finish reason:", response.choices[0].finish_reason)
print("\nParsed:")
if raw is None:
    # Not a parse failure — the model produced no content at all. A token cap
    # shows up as finish_reason "length"; a refusal comes back as "stop" with
    # the reason only in message.refusal. Print both or you will misread it.
    print("No content. refusal:", response.choices[0].message.refusal)
else:
    try:
        print(json.loads(raw))
    except json.JSONDecodeError as e:
        print(f"JSON parse failed: {e}")

Notice what is deliberately missing: there is no response_format here at all, even though classify_record sends a strict JSON schema. That is the point of the script. The schema is a constraint applied to the output, and the stronger it is, the more it hides — with strict: True the API will not let the model wrap its answer in markdown fences, add a preamble, or name a fifth category, so you never find out that it wanted to. Dropping the schema shows you what the model actually reaches for when nothing stops it. If a classification is consistently odd, that unconstrained answer usually tells you the prompt is at fault, not the parser.

This lets you see exactly what the model returned, without your validation code masking it.

Build activity

1. Create debug_run.py and run it against your most recent pipeline_output.json. Verify the stage gaps are all zero (or explain any gap you see).
2. Deliberately introduce a failure: in your CSV, change one record's date field to a format the validator does not recognise (e.g. "January 8 2024"). Run the pipeline and use debug_run.py to locate the failing stage.
3. Filter the JSONL log for WARNING entries in the validate stage to identify the exact failing record.
4. Create isolate_validate.py and reproduce the failure for that single record in isolation.
5. Fix the date validator in schema.py to handle the new format. Re-run isolate_validate.py to confirm the fix.
6. Re-run the full pipeline with the updated validator. Verify all 30 records pass validation.

Challenge Chase

Your pipeline currently halts when a stage produces a critical failure (the state.halted flag). But halting loses all the work done before the failure point. Design a checkpoint-and-resume system: after each stage completes successfully, serialise the current PipelineState to a JSON file. If the pipeline halts at stage 4, resuming should reload the state from the stage 3 checkpoint and continue from there — without re-running stages 1 to 3. What are the failure modes of this approach? When would checkpoint state be invalid or dangerous to resume from?

The code from this lesson

The four debugging scripts from this lesson, alongside the pipeline they inspect. Nothing in the pipeline itself changed here — this lesson adds tools, not stages.

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

• I can name the four classes of pipeline failure and describe where each appears
• debug_run.py reads pipeline_output.json and prints the records_in/out gap for every stage
• I have deliberately introduced a validation failure and used the structured log to locate it
• I have reproduced the failure in isolation using a minimal PipelineState
• I have fixed the root cause (not patched the symptom) and verified the fix
• I can explain why swallowing exceptions and skipping failing records is dangerous in a compliance pipeline

Quick Check

KSB evidence focus

• K11 — Understands relevant data governance, data privacy and security issues. A systematic debugging approach protects data integrity. A pipeline that silently drops records — or routes them to the wrong queue — may create compliance gaps that are not visible until an audit. The framework you have built ensures that every record is accounted for at every stage.

• S10 — Can carry out data processing tasks, including processing/cleaning, data transformations, and feature engineering. Debugging a pipeline failure is a data processing task: you read structured log data, filter it by stage and level, and use it to reconstruct what happened to specific records. The query patterns you wrote in L3.4 are the tools you use here.

• S11 — Can design, implement and test an AI/ML model or system. The ability to isolate a single stage and reproduce a failure in isolation is a hallmark of testable system design. If you had written the pipeline as one large script, this would not be possible. The modular stage architecture you built in Unit 1 is what makes systematic debugging tractable.

Up next: Unit 4 puts everything together. You will assemble the complete pipeline, measure its performance, optimise the bottlenecks, and write the technical narrative that explains every design decision to your portfolio assessors.