The Most Important Decision in Your Pipeline: Workflow or Agent?

Module 5, Unit 2 | Lesson 1 of 3

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

• Apply a three-question framework to decide whether a pipeline stage should use deterministic workflow logic or an LLM agent (K11, K12)
• Implement a real workflow function with explicit rules, typed inputs and outputs, and unit tests (S10, B6)
• Explain why testability is the key practical difference between workflows and agents, and why that matters for production systems (K11, S10)
• Identify which stages of your pipeline are workflow-appropriate and which require agentic reasoning (K12, S10, B6)

Unit 1 gave you a working pipeline skeleton with five named stages. Now you need to decide, stage by stage, what kind of logic goes inside each one. This is not a philosophical question — it is an engineering decision with real consequences for cost, reliability, and debuggability.

The core choice is this: workflow or agent?

🔑 Key term — Deterministic workflow: A processing step whose output is fully determined by its input and the rules you have written. Given the same data, it always produces the same result. It contains no LLM calls. It can be tested with unit tests, audited completely, and explained line by line.

🔑 Key term — LLM agent: A processing step that sends data to a language model and uses the model's response to make a decision or produce output. Given the same data, it may produce different results on different runs. It can handle ambiguity and natural language that no explicit rule can capture — but it costs money per call, can be wrong, and is harder to test.

The decision framework

Ask three questions about every stage in your pipeline:

1. Can the logic be expressed as explicit rules?

If yes — use a workflow. Explicit rules include: normalise tool names, check whether cost exceeds a threshold, convert date formats, route pii_detected records to the compliance queue. These tasks require matching and comparison, not understanding.

2. Does the task involve ambiguity or natural language interpretation?

If yes — consider an agent. Deciding whether a free-text incident description represents a compliance risk, or whether an unusual usage pattern is suspicious, requires reading and understanding — not pattern matching.

3. Does correctness matter more than nuance?

If yes — use a workflow. A routing rule that sends pii_detected records to the compliance queue must always route correctly. An LLM might classify the same record differently on different runs. That 3% inconsistency is acceptable for a category label; it is unacceptable for a compliance decision.

Coach Cora

The most common mistake in AI pipeline design is reaching for an agent when a workflow would do. LLMs are expensive, slow relative to code, and produce different results on different runs. Every time you replace an if statement with an LLM call, you are accepting more cost, more latency, and more failure modes. Only use an agent when the task genuinely requires understanding that code cannot provide.

Apply the framework to your pipeline

Three stages are workflow-appropriate. One stage requires LLM reasoning — classify. route uses the classification as input, but the routing rule itself (pii_detected → compliance queue) is deterministic; it does not call an LLM.

Build the improved ingest stage

The placeholder stage_ingest from L1.3 works, but a production version handles edge cases. Update it in stages.py:

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

    with open(state.source_file, newline="", encoding="utf-8") as f:
        reader = csv.DictReader(f)
        for i, row in enumerate(reader, start=1):
            # Skip entirely empty rows. A row with fewer fields than the
            # header gives None rather than "", so guard before .strip().
            if all((v or "").strip() == "" for v in row.values()):
                skipped += 1
                continue
            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=skipped,
        duration_seconds=round(time.time() - started, 3),
        metadata={
            "source_file": state.source_file,
            "skipped_empty_rows": skipped,
        },
    ))

    print(f"[ingest]   {len(rows)} rows read, {skipped} empty rows skipped")
    return state

The _source_row field carries the original CSV line number through the pipeline. When a record fails validation, you can report exactly which row in the source file caused it.

Build a testable workflow function

Looking at the data, some records have a high cost_usd but the flagged_issue field still says none — the collection layer missed them. Add a workflow function to stages.py that catches these. It stays unwired from the pipeline for the rest of the module, deliberately: whether such records should be re-flagged is the data owner's call, not a validator's. It is here to be tested.

from schema import FlaggedIssue


def workflow_flag_cost_overruns(
    records: list[ToolUsageRecord],
    cost_threshold: float = 0.10,
) -> list[tuple[ToolUsageRecord, bool]]:
    """
    Returns (record, is_newly_flagged) pairs.
    Flags records where cost_usd exceeds the threshold
    AND flagged_issue is currently 'none'.
    Does not modify records — flags are applied downstream.
    """
    results = []
    for record in records:
        newly_flagged = (
            record.cost_usd > cost_threshold
            and record.flagged_issue == FlaggedIssue.NONE
        )
        results.append((record, newly_flagged))

    flagged_count = sum(1 for _, f in results if f)
    print(
        f"[workflow] cost_overrun check: {flagged_count} newly flagged "
        f"(threshold: ${cost_threshold:.2f})"
    )
    return results

This function is pure workflow logic — deterministic, no LLM calls, and completely testable.

Keep the function and its tests — they are part of your L4.3 submission, and "I built it, tested it, and deliberately did not enable it" is a stronger answer than either wiring it in or leaving it out.

Write the unit tests

Create test_workflow.py:

from datetime import date
from schema import ToolUsageRecord, ToolName, FlaggedIssue
from stages import workflow_flag_cost_overruns


def make_record(cost_usd: float, flagged_issue: str = "none") -> ToolUsageRecord:
    return ToolUsageRecord(
        log_id="LOG-TEST",
        date=date(2024, 1, 8),
        learner_id="L001",
        tool_name="GPT-4",
        task_type="summarisation",
        model_used="gpt-4-turbo",
        prompt_tokens=100,
        completion_tokens=50,
        response_time_ms=1000,
        quality_score=4.0,
        time_saved_mins=10,
        flagged_issue=flagged_issue,
        human_reviewed=True,
        cost_usd=cost_usd,
    )

Then the tests themselves, and the runner beneath them. Each test is three or four lines because the factory above absorbed the setup — that is the whole reason it exists.

def test_flags_high_cost_unflagged_record():
    results = workflow_flag_cost_overruns([make_record(0.15)])
    assert results[0][1] is True, "Should flag high-cost unflagged record"


def test_does_not_flag_already_flagged_record():
    results = workflow_flag_cost_overruns(
        [make_record(0.15, flagged_issue="cost_overrun")]
    )
    assert results[0][1] is False, "Should not double-flag an already-flagged record"


def test_does_not_flag_low_cost_record():
    results = workflow_flag_cost_overruns([make_record(0.05)])
    assert results[0][1] is False, "Should not flag a low-cost record"


if __name__ == "__main__":
    test_flags_high_cost_unflagged_record()
    test_does_not_flag_already_flagged_record()
    test_does_not_flag_low_cost_record()
    print("All tests passed.")

Run it two ways. First, directly — the __main__ block at the bottom means the file works as a plain script with no extra tooling:

python test_workflow.py
# [workflow] cost_overrun check: 1 newly flagged (threshold: $0.10)
# [workflow] cost_overrun check: 0 newly flagged (threshold: $0.10)
# [workflow] cost_overrun check: 0 newly flagged (threshold: $0.10)
# All tests passed.

One [workflow] line per test, because the function prints every time it runs — one flagged in the first test, none in the other two, which is exactly what the three assertions say. Under pytest you will not see them: pytest captures stdout and only shows it for a test that fails.

Then with pytest, the standard Python test runner. Add it to requirements.txt, which now reads:

pandas
pydantic>=2
openai>=3
requests
python-dotenv
tiktoken
pytest

And install it:

pip install -r requirements.txt
python -m pytest test_workflow.py -v

test_workflow.py::test_flags_high_cost_unflagged_record PASSED
test_workflow.py::test_does_not_flag_already_flagged_record PASSED
test_workflow.py::test_does_not_flag_low_cost_record PASSED

Pytest discovers anything named test_* automatically, reports each test separately, and — unlike the script form — keeps going after the first failure so you see every problem in one run. That difference stops mattering with three tests and starts mattering a lot with thirty. Use pytest from here on; the __main__ block stays as a convenience.

All three tests should pass either way. This is what testable workflow logic looks like.

Notice that make_record passes a real date(2024, 1, 8) object rather than the string "2024-01-08". That works because the parse_date validator you wrote in L1.2 returns early when it is already handed a date. This is the first payoff from writing that guard: the schema accepts data from the CSV and from a test fixture, without the test having to pretend to be a CSV.

Curious Cat

You cannot write this kind of test for an LLM agent. Given the same record, the agent might classify it as cost_anomaly today and operational_issue tomorrow. You can test that it returns something in the right format — but you cannot assert the exact output. That unpredictability is fine for classification; it is unacceptable for a rule that determines whether a record goes to compliance. This is why the framework matters: put deterministic logic in workflows, and reserve agents for the decisions that genuinely require flexibility.

The agent stub

For contrast, here is what the agent stage will eventually look like. For now it is a well-documented stub — it defines the interface without making any real API calls:

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

    Records with flagged_issue == 'none' pass through without an LLM call.
    Flagged records are classified into one of:
        - compliance_risk
        - operational_issue
        - cost_anomaly
        - data_quality

    Unit 3 (L3.1) implements the real classifier.
    The interface — what goes in, what comes out — stays the same.
    """
    import time
    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": "stub — real agent added in Unit 3"},
    ))

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

The docstring is the spec. When you implement the real classifier in L3.1, you will replace the stub body while keeping the interface — what goes in and what comes out — exactly the same.

Build activity

1. Update stage_ingest in stages.py with the improved version above (empty row skipping and _source_row tracking).
2. Add workflow_flag_cost_overruns to stages.py.
3. Write test_workflow.py and run it with python -m pytest test_workflow.py -v — all three tests must pass.
4. Update stage_classify in stages.py with the documented stub above.
5. Run python pipeline.py and confirm all five stages complete cleanly.

Challenge Chase

workflow_flag_cost_overruns uses a hardcoded default threshold of $0.10. In a real deployment, the threshold might differ by task type — code generation might justify $0.12, summarisation only $0.05. Design a ThresholdConfig dataclass that holds per-task-type thresholds, and update the function to look up the right threshold for each record. What happens to records whose task_type is not in the config — should they default to the global threshold, or be flagged for review?

The code from this lesson

The pipeline as this lesson leaves it: a real ingest stage, the tested workflow function, and the agent stub waiting for Unit 3. 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

• I can apply the three-question framework to any pipeline stage and justify my workflow-vs-agent decision
• The improved stage_ingest tracks _source_row and skips empty rows
• workflow_flag_cost_overruns is in stages.py and correctly identifies newly-flagged records
• All three unit tests in test_workflow.py pass under python -m pytest test_workflow.py -v
• stage_classify has a clear docstring describing its intended interface, even though the body is a stub
• python pipeline.py runs to completion with all five stages

Quick Check

KSB evidence focus

• K11 — Understands relevant data governance, data privacy and security issues. The workflow-vs-agent decision is a data governance decision. Deterministic routing of compliance records is not optional — it must be consistent, auditable, and explainable. Your decision framework documents why each stage was implemented the way it was, which is the kind of reasoning a data governance audit would expect to find.

• K12 — Understands how to set up, interact with and generate APIs, databases and spreadsheets. The agent stub defines the interface your pipeline will use to call an LLM API in Unit 3. Writing the interface before the implementation is a professional software engineering practice — it lets the rest of the pipeline be built and tested without waiting for the LLM integration to be complete.

• S10 — Can carry out data processing tasks, including processing/cleaning, data transformations, and feature engineering. workflow_flag_cost_overruns is a data processing function: it reads typed records, applies a business rule, and produces a labelled output. It is fully tested and documented. That is the standard for production data processing code.

Up next: Lesson 2 goes deeper into the mechanics of wiring stages together. You will implement real conditional branching in stage_route, add parallel processing for independent record operations, and write retry logic for the API calls that are coming in Unit 3.