Making Every Decision Traceable: Structured Logging

Module 5, Unit 3 | Lesson 4 of 5

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

• Explain why print statements are not sufficient for production pipeline observability (K11, B6)
• Design a structured log schema that captures every pipeline decision in a queryable format (K11, S10)
• Implement a Python logger that writes JSONL log entries at key points across all pipeline stages (K12, S10)
• Query your structured logs to reconstruct the history of any individual record through the pipeline (S10, S11)

Your pipeline is now live — it ingests, validates, classifies, routes, approves, and outputs. But when a stakeholder asks "why was LOG-017 routed to ops_review?" or when a pipeline run produces unexpected results at 3am, can you answer? If your only visibility is the print statements in each stage, the answer is no — they have already scrolled off the terminal and are gone.

Structured logging is the mechanism that makes your pipeline observable: every decision is written to a persistent, queryable file. You can reconstruct exactly what happened to any record, in any run, at any time.

🔑 Key term — Structured logging: Writing log entries as machine-readable data (JSON) rather than human-readable strings. Structured logs can be queried, filtered, and aggregated programmatically. A plain print("[route] LOG-017 → ops_review") cannot be searched or filtered efficiently. A JSON entry {"event": "route_decision", "log_id": "LOG-017", "route": "ops_review", "reason": "hallucination"} can.

Why not just use print?

Print statements are fine for development. They are not fine for production because:

• They vanish when the terminal closes
• They cannot be filtered by level (info vs warning vs error)
• They cannot be queried — you cannot ask "show me all records where confidence was below 0.75"
• They provide no timestamps — you cannot tell when something happened relative to other events
• They carry no context — you cannot tell which run, which stage, or which record generated the message

Curious Cat

JSONL (JSON Lines) is a file format where each line is a complete, valid JSON object. It is the standard format for structured log files because it is easy to write (append one line per event), easy to read line by line without loading the whole file, and trivially parseable with any JSON library. Every major observability platform — Datadog, Splunk, Elastic — ingests JSONL natively.

Those are general-purpose log stores. There is also a category of tool built specifically for LLM pipelines — LangSmith, Langfuse and Arize Phoenix are the ones you will meet most often, and OpenTelemetry now publishes GenAI semantic conventions that describe the same data in a vendor-neutral way. What they add over a log file is per-call token and cost attribution, prompt and model versioning so you can tell which version of a prompt produced which answer, and nesting: a run becomes a trace, each stage a span, each model call a child span with its inputs and outputs attached.

The schema you are about to design maps onto that almost directly — run_id is the trace identifier, stage is the span, record_id is the attribute you filter on — which is the reason to build it by hand once before reaching for a platform. Adopt one when you have more pipelines than you can reasonably grep, or when someone other than you needs to answer "what did this pipeline do to this record". Until then a file you completely understand is worth more than a dashboard you do not, and you now know what the dashboard would be doing.

Design the log entry schema

Before writing any code, decide what every log entry needs to contain:

{
  "timestamp":  ISO-8601 datetime string
  "run_id":     UUID from PipelineState — ties all entries for one run together
  "stage":      which stage generated the entry
  "event":      what happened (stage_start, record_classified, record_routed, ...)
  "level":      INFO | WARNING | ERROR
  "log_id":     record identifier, if applicable (null for stage-level events)
  "data":       event-specific payload (arbitrary dict)
}

The run_id is the key field: it lets you pull all log entries for a single pipeline run in one query, even if multiple runs are interleaved in the same log file.

Build the logger

Create pipeline_logger.py:

from __future__ import annotations

import json
import os
import threading
from datetime import datetime, timezone
from typing import Any


class PipelineLogger:
    """
    Structured logger that writes JSONL entries to a persistent log file.
    Each entry is a complete JSON object on its own line.
    The run_id ties all entries for one pipeline run together.
    """

    def __init__(self, log_path: str, run_id: str) -> None:
        self.log_path = log_path
        self.run_id = run_id
        # stage_classify logs from several worker threads at once
        self._lock = threading.Lock()
        # Ensure log directory exists
        os.makedirs(os.path.dirname(log_path) or ".", exist_ok=True)

One private method does all the work, and it is the only place that touches the file. The lock matters: stage_classify runs its workers in parallel, and two threads appending to the same file without one produce interleaved half-lines.

    def _write(
        self,
        stage: str,
        event: str,
        level: str = "INFO",
        log_id: str | None = None,
        data: dict[str, Any] | None = None,
    ) -> None:
        entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "run_id": self.run_id,
            "stage": stage,
            "event": event,
            "level": level,
            "log_id": log_id,
            "data": data or {},
        }
        # default=str so a stray date or Decimal in `data` degrades to a string
        # rather than killing the run with a TypeError mid-pipeline.
        line = json.dumps(entry, default=str) + "\n"
        with self._lock:
            with open(self.log_path, "a", encoding="utf-8") as f:
                f.write(line)

The three public methods are thin wrappers over _write, differing only in level. Keeping them separate means a call site reads as logger.warning(...) rather than passing a level string around.

    def info(
        self,
        stage: str,
        event: str,
        log_id: str | None = None,
        **data: Any,
    ) -> None:
        self._write(stage, event, "INFO", log_id, data)

    def warning(
        self,
        stage: str,
        event: str,
        log_id: str | None = None,
        **data: Any,
    ) -> None:
        self._write(stage, event, "WARNING", log_id, data)

    def error(
        self,
        stage: str,
        event: str,
        log_id: str | None = None,
        **data: Any,
    ) -> None:
        self._write(stage, event, "ERROR", log_id, data)

Add the logger to PipelineState

The logger needs to travel through all stages as part of the pipeline state. Add two fields to PipelineState in contracts.py — invalid_rows and logger, the last two below. Here is the whole dataclass so nothing gets lost, including the four queues from L2.2:

from pipeline_logger import PipelineLogger


@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 = ""
    human_review_queue: list[dict[str, Any]] = field(default_factory=list)
    compliance_queue: list[dict[str, Any]] = field(default_factory=list)
    ops_queue: list[dict[str, Any]] = field(default_factory=list)
    standard_queue: list[dict[str, Any]] = field(default_factory=list)
    invalid_rows: list[dict[str, Any]] = field(default_factory=list)
    logger: PipelineLogger | None = None

invalid_rows gives failed records somewhere to live on the state itself. Until now a row that failed validation existed only inside StageResult.errors; promoting it onto PipelineState is what lets the isolation tools you write in L3.5 read it directly.

Initialise the logger in pipeline.py when the run starts. One log file per run, named after the run_id, so two runs can never interleave in the same file:

from pipeline_logger import PipelineLogger


def run_pipeline(source_file: str, connector=None) -> PipelineState:
    if connector is None:
        connector = JSONFileConnector("pipeline_output.json")

    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,
        ),
    )

    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

    if not state.halted:
        state = stage_output(state, connector=connector)

    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")

The run_id is truncated to eight characters here — long enough to be unique across your runs, short enough to type when you are hunting for a log file.

Add log entries to each stage

Log at the start of each stage, at key record-level decisions, and at errors. stage_validate is the worked example, and it splits into two functions: the per-row loop, which is where the record-level logging lives, and the stage itself.

Replace your L1.3 version with both. The failure accumulator is now called invalid and holds a different shape, and it is also stored on state.invalid_rows; the halt check is otherwise unchanged from L1.3. The StageResult append does change, in a way worth pausing on: invalid keeps the whole rejected row, because that is what the isolation tools in L3.5 need, but only a projection of it is handed to the StageResult. Anything in there is published by run_metadata in L2.3, and a row that failed validation can hold the same personal data this pipeline otherwise keeps on the machine. Internal state and published output are different audiences.

def validate_rows(
    state: PipelineState,
) -> tuple[list[ToolUsageRecord], list[dict[str, Any]]]:
    """Validate every raw row, logging each decision. Returns (valid, invalid)."""
    valid: list[ToolUsageRecord] = []
    invalid: list[dict[str, Any]] = []

    for row in state.raw_rows:
        log_id = row.get("log_id", "UNKNOWN")
        try:
            record = ToolUsageRecord.model_validate(row)
            valid.append(record)
            state.logger.info(
                "validate", "record_valid", log_id=log_id,
                tool_name=row.get("tool_name"),
                flagged_issue=row.get("flagged_issue"),
            )
        except ValidationError as e:
            errors = [err["msg"] for err in e.errors()]
            invalid.append({
                "log_id": log_id,
                "source_row": row.get("_source_row"),
                "row": row,
                "errors": errors,
            })
            state.logger.warning(
                "validate", "record_invalid", log_id=log_id,
                errors=errors,
                source_row=row.get("_source_row"),
            )

    return valid, invalid

ToolUsageRecord.model_validate(row) replaces the dict comprehension from L1.3 — Pydantic ignores the extra _source_row key by default, so filtering it out by hand was never necessary.

The stage keeps the shape you already know. What is new is that the halt now writes an ERROR entry to the log as well as printing it, so a halted run leaves a permanent record even though — since L2.3 — it writes no output file:

def stage_validate(state: PipelineState) -> PipelineState:
    """Stage 2: Validate and normalise raw rows, logging every decision."""
    started = time.time()
    state.logger.info("validate", "stage_start", records_in=len(state.raw_rows))

    valid, invalid = validate_rows(state)
    state.valid_records = valid
    state.invalid_rows = invalid

    # Halt if error rate exceeds threshold — unchanged from L1.3
    error_rate = len(invalid) / 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(invalid)} of {len(state.raw_rows)} records failed)"
        )
        state.logger.error(
            "validate", "halt_threshold_exceeded",
            error_rate=round(error_rate, 3),
            reason=state.halt_reason,
        )

    state.stage_results.append(StageResult(
        stage_name="validate",
        records_in=len(state.raw_rows),
        records_out=len(valid),
        records_failed=len(invalid),
        duration_seconds=round(time.time() - started, 3),
        # `invalid` keeps the whole rejected row for the isolation tools in
        # L3.5. What gets *published* is narrower: a rejected row can carry the
        # very PII this pipeline routes to a local model, and this list travels
        # to whatever the connector points at. Name the failure, not its
        # contents.
        errors=[
            {
                "log_id": e["log_id"],
                "source_row": e["source_row"],
                "errors": e["errors"],
            }
            for e in invalid
        ],
    ))

    state.logger.info(
        "validate", "stage_complete",
        valid=len(valid), invalid=len(invalid),
        duration_s=round(time.time() - started, 3),
    )

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

    return state

Every stage now leaves two traces: one in the run summary that pipeline_output.json carries, and one in the log that survives a halt.

Add equivalent entries for every record the classify stage touched — the ones it classified and the ones it could not. They go in stage_classify, in two loops after run_in_parallel returns, not inside classify_item. The reason is the one you already met when counting: classify_item runs on several worker threads at once, and a single logger writing from all of them is one more piece of shared state to get right. Logging from the results afterwards costs nothing and cannot interleave:

    # stages.py — inside stage_classify(), after run_in_parallel returns
    for r in results:
        state.logger.info(
            "classify", "record_classified", log_id=r["record"].log_id,
            classification=r["classification"],
            confidence=r["confidence"],
            model=r["model_used"],
            route_override=r.get("route_override"),
        )
    for e in errs:
        state.logger.error(
            "classify", "record_classification_failed",
            log_id=e["item"]["record"].log_id,
            error=e["error"],
        )

The second loop is the one people forget. run_in_parallel returns records it could not process in errs, not in results, so a record whose classifier call failed every retry is absent from the first loop entirely. Log only the successes and the record leaves no trace anywhere: not in the output, because it never reached routing; not in the log, because it never reached results. The count in records_failed would be the only sign it existed.

stage_route gets the same treatment — a loop over routed, after the queues are filled:

    # stages.py — inside stage_route(), after routed is built
    for item in routed:
        state.logger.info(
            "route", "record_routed", log_id=item["record"].log_id,
            route=item["route"],
            flagged_issue=item["record"].flagged_issue.value,
        )

Coach Cora

Log at decisions, not at every line of code. Too much logging creates noise that is as useless as no logging — you have to find the signal in your own output. The principle is: log when a record's status changes (valid/invalid, classified, routed), when something unexpected happens (validation error, retry, halt), and at stage boundaries (start and complete). Everything else is development debugging that should be removed before production.

Query your structured logs

After running the pipeline, your logs/ folder contains pipeline_{run_id}.jsonl. You can query it in Python:

# query_logs.py
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]

def load_log(path: str) -> list[dict]:
    entries = []
    with open(path, encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if line:
                entries.append(json.loads(line))
    return entries

Two queries. The first answers the question you will actually ask of a log — everything that happened to one record, in order. The second sweeps for the low-confidence classifications that the L3.1 threshold routed to human review.

def trace_record(entries: list[dict], log_id: str) -> list[dict]:
    """Return all log entries for a specific record."""
    return [e for e in entries if e.get("log_id") == log_id]


def find_low_confidence(entries: list[dict], threshold: float = 0.75) -> list[dict]:
    """Find all classification events where confidence was below threshold."""
    return [
        e for e in entries
        if e.get("event") == "record_classified"
        and e.get("data", {}).get("confidence", 1.0) < threshold
    ]

And an aggregate: how many records went down each route. Three functions over one JSONL file is the whole of what a log platform gives you at this scale.

def count_by_route(entries: list[dict]) -> dict:
    """Count records sent to each route."""
    from collections import Counter
    routes = [
        e["data"]["route"]
        for e in entries
        if e.get("event") == "record_routed"
    ]
    return dict(Counter(routes))


if __name__ == "__main__":
    log = load_log(LOG_FILE)
    print(f"Total entries: {len(log)}")
    print("\nTrace LOG-017:")
    for e in trace_record(log, "LOG-017"):
        print(f"  [{e['stage']}] {e['event']} - {e['data']}")
    print("\nLow-confidence classifications:")
    for e in find_low_confidence(log):
        print(f"  {e['log_id']}: {e['data']}")
    print("\nRoute counts:", count_by_route(log))

Build activity

1. Create pipeline_logger.py with the PipelineLogger class.
2. Add logger: PipelineLogger | None = None to PipelineState in contracts.py.
3. Initialise the logger in pipeline.py when creating PipelineState, writing to logs/pipeline_{run_id}.jsonl.
4. Add structured log entries to stage_validate, stage_classify, and stage_route at the points shown above.
5. Log the run lifecycle in pipeline.py, not just the stages: run_start before the first stage, run_complete after the last, and — inside the halt branch — an ERROR-level halted entry carrying state.halt_reason. Without that last one a halted run leaves no machine-readable trace at all, which is precisely the run you will most want to query later.
6. Run python pipeline.py and open the generated JSONL file. Verify there is one entry per validation event and one per classification, bracketed by run_start and run_complete.
7. Write query_logs.py and run it. Trace a specific record through the pipeline using trace_record.

Challenge Chase

Your current logger writes to a single file per run. For a pipeline that runs hundreds of times, this creates hundreds of small files. Design a LogAggregator class that reads multiple JSONL files from a logs/ directory and merges them into a single queryable dataset, filtering by date range, run_id, or record id. How would you handle a query that spans 30 days of pipeline runs efficiently — without loading all 30 files into memory at once?

The code from this lesson

The pipeline with structured logging and the query helpers. The two per-record log loops are shown above as patches rather than whole functions, so they are not in the archive; stage-level logging is.

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

• pipeline_logger.py exists with PipelineLogger and info, warning, error methods
• _write serialises the entry before taking the lock, and holds the lock only around the file append
• PipelineState has a logger field initialised in pipeline.py
• stage_validate logs stage_start, record_valid, record_invalid, and stage_complete
• stage_classify logs record_classified with classification, confidence, and model
• stage_route logs record_routed with the route and flagged_issue
• Running the pipeline generates a JSONL file in logs/
• query_logs.py can trace a single record and count records by route

Quick Check

KSB evidence focus

• K11 — Understands relevant data governance, data privacy and security issues. A pipeline that processes compliance-relevant records must be auditable: you need to prove what happened to each record, when, and why. Structured logging is the technical mechanism that makes this possible. Without it, your pipeline makes decisions that cannot be reviewed or challenged.

• S10 — Can carry out data processing tasks, including processing/cleaning, data transformations, and feature engineering. query_logs.py demonstrates data processing applied to pipeline telemetry: reading JSONL, filtering by field values, aggregating by category. The same techniques you used to profile the raw dataset in L1.1 apply here to pipeline observability data.

• S11 — Can design, implement and test an AI/ML model or system. An observable system is a testable system. Being able to trace any record through all stages, see the exact confidence score and reasoning, and query across runs means you can verify your pipeline is behaving correctly — not just that it is running.

Up next: Logging tells you what happened. Lesson 5 shows you how to read those logs when something goes wrong — a systematic approach to debugging AI pipeline failures, from validation errors through to classifier misbehaviour.