How to Build a Multi-Stage AI Agent Pipeline in Python to Stop Errors From Compounding

Most tutorials about AI agents show you a single model call, or a single agent that loops back on itself until it is satisfied with its own answer. Real systems that use AI agents for document-heavy work rarely look like that. They look more like an assembly line: one step reads a document and pulls out structured facts, the next step does some math, a third step makes a decision from those facts, and a fourth step checks the decision before anyone trusts it. Red Hat’s own AI quickstart catalog, for example, ships a ten-stage agent pipeline that reads aircraft lease contracts, reconciles them against flight-hour and maintenance data, and flags variances for a human to review, one agent per stage, chained together with a graph orchestration library.

In this tutorial you will build a small version of that same shape of system from scratch, using nothing but Python and a model running locally through Ollama. The scenario is a Service Level Agreement (SLA) credit reconciliation pipeline: read a vendor contract, compute how much the service was actually down last month, decide what credit the customer is owed, and check that decision before it goes anywhere near an invoice. Along the way you will watch the pipeline get things wrong in several genuinely different ways, at every single stage, using nothing but a small, honest, unmodified 1.5-billion-parameter model. You will also watch a deterministic validation gate catch every one of those mistakes before they could have cost anyone real money.

Before the first command, here is what a few terms mean in plain language:

• A pipeline stage here is just a Python function. Some stages call a language model; others are plain arithmetic. Nothing loops back, the data flows in one direction from stage 1 to stage 4.
• Structured output means asking the model to return JSON that matches a schema you define, instead of free-form prose you would have to parse yourself.
• A validation gate is a stage that does not call a model at all. It independently recomputes what the answer should be and compares that against what an earlier stage produced, refusing to pass a mismatch downstream.

What you need before you start

• Windows, macOS, or Linux. This tutorial was written and tested on Windows with Python 3.13.14; any Python 3.10 or newer works the same way.
• Ollama installed and running, with the qwen2.5:1.5b model pulled (ollama pull qwen2.5:1.5b, about 986 MB). Everything in this tutorial runs on CPU and costs nothing per request.
• Comfort with basic Python: functions, dataclasses, dictionaries. No prior agent-framework experience is assumed, and this tutorial does not use LangGraph, CrewAI, or any other orchestration library. The pipeline is four plain functions calling each other in order.
• About 10 minutes of wall-clock time to run every step, since the model calls are the slow part.

Step 1: Set up the project

Keep this tutorial’s dependencies isolated in their own virtual environment:

# Context: any terminal, in a folder you keep coding projects in.
# Purpose: create a project folder and an isolated Python virtual environment.
mkdir sla-pipeline
cd sla-pipeline
python -m venv venv
# Windows: venv\Scripts\activate
# macOS/Linux: source venv/bin/activate
pip install ollama pytest

Expected output: pip resolves and installs ollama and pytest, ending in a line starting Successfully installed. This tutorial was built against ollama 0.6.2 (the Python client library, not to be confused with the Ollama application itself) and pytest 9.1.1.

Step 2: Define the scenario

The pipeline reconciles a monthly SLA credit from two inputs: a contract that defines uptime commitments and credit tiers, and an incident log recording every outage that month. Create contract_a.txt:

NORTHWIND CLOUD SERVICES: SERVICE LEVEL AGREEMENT (EXHIBIT C)

1. Monthly Uptime Commitment. Northwind will use commercially reasonable efforts
to make the Service available with a Monthly Uptime Percentage of at least 99.9%
during any calendar month.

2. Service Credits. If the Monthly Uptime Percentage for a calendar month falls
below the commitment above, Customer will be eligible for a Service Credit
according to the following schedule:

   - Monthly Uptime Percentage below 99.9% but at or above 99.0%: Service
     Credit equal to 10% of the fees paid for that month.
   - Monthly Uptime Percentage below 99.0% but at or above 95.0%: Service
     Credit equal to 25% of the fees paid for that month.
   - Monthly Uptime Percentage below 95.0%: Service Credit equal to 50% of the
     fees paid for that month.

3. Excused Downtime. Any unavailability that occurs during a maintenance
window that Northwind announced to Customer at least 24 hours in advance
("Scheduled Maintenance") is excused and will not count against the Monthly
Uptime Percentage.

This is a completely standard three-tier SLA credit schedule, the same shape used by most SaaS vendors. Now create incidents_april_2026.csv, a downtime log for the month being reconciled:

date,start,end,minutes,category,description
2026-04-03,02:00,04:00,120,scheduled_maintenance,Database failover test; announced 2026-04-01 via status page
2026-04-10,14:22,14:58,36,unplanned,Load balancer misconfiguration during a routine deploy
2026-04-17,08:05,08:12,7,unplanned,Upstream DNS provider incident
2026-04-22,20:00,23:00,180,scheduled_maintenance,Platform migration; announced 2026-04-18 via status page
2026-04-28,03:00,04:47,107,unplanned,Primary datacenter power event

Two of these five incidents are scheduled maintenance, announced with proper notice per Section 3 of the contract, and should not count against uptime. The other three are unplanned and should. This is a deliberately chosen dataset: the correct answer is fully computable by hand, which is exactly what makes it useful for checking whether an AI pipeline gets it right.

Step 3: Why not just ask a model to do the whole thing in one prompt?

Before building four separate stages, it is worth asking whether you need more than one. Create one_shot_comparison.py and hand the model both files at once, asking it to compute the uptime and pick a credit tier in a single call:

import json
import ollama

SCHEMA = {
    "type": "object",
    "properties": {
        "monthly_uptime_pct": {"type": "number"},
        "selected_credit_pct": {"type": ["number", "null"]},
        "justification": {"type": "string"},
    },
    "required": ["monthly_uptime_pct", "selected_credit_pct", "justification"],
}

PROMPT = """Read the SLA contract and the incident log below. Compute the
Monthly Uptime Percentage for the month (remember to exclude any downtime
the contract says is excused), then decide which credit tier applies and
why.

CONTRACT:
---
{contract_text}
---

INCIDENT LOG (CSV, minutes column is downtime in minutes for that incident):
---
{incident_csv}
---

The month has {days_in_month} days. Return only the JSON object, nothing else.
"""

def one_shot(contract_path, incidents_path, days_in_month, model="qwen2.5:1.5b"):
    with open(contract_path, encoding="utf-8") as f:
        contract_text = f.read()
    with open(incidents_path, encoding="utf-8") as f:
        incident_csv = f.read()
    response = ollama.chat(
        model=model,
        messages=[{"role": "user", "content": PROMPT.format(
            contract_text=contract_text, incident_csv=incident_csv, days_in_month=days_in_month)}],
        format=SCHEMA,
        options={"temperature": 0},
    )
    return json.loads(response["message"]["content"])

if __name__ == "__main__":
    for contract in ["contract_a.txt", "contract_b.txt"]:
        print(f"=== one-shot: {contract} ===")
        print(json.dumps(one_shot(contract, "incidents_april_2026.csv", 30), indent=2))

(Ignore the reference to contract_b.txt for now, you will create it in Step 5. Run it once with just contract_a.txt if you want to see the result immediately.) The correct answer, worked out by hand: 30 days is 43,200 minutes; the three unplanned incidents total 150 minutes of counted downtime; 43,050 / 43,200 is 99.6528% uptime, which lands in the first tier for a 10% credit. Here is what qwen2.5:1.5b actually returned, captured directly from a real run:

=== one-shot: contract_a.txt ===
{
  "monthly_uptime_pct": 96.5,
  "selected_credit_pct": 25,
  "justification": "The incident that caused excused downtime (scheduled maintenance) is excluded from the calculation of Monthly Uptime Percentage. The remaining incidents are below the minimum commitment, so Customer is eligible for a Service Credit equal to 25% of the fees paid for that month."
}

Both numbers are wrong: 96.5% instead of 99.6528%, and a 25% credit tier instead of 10%. If the vendor’s monthly fee were $40,000, that is the difference between a correct $4,000 credit and an incorrect $10,000 credit, a $6,000 overpayment based on a percentage the model appears to have invented rather than computed. Asking a model to read a CSV, exclude specific rows by category, sum the rest, and divide by a total is asking it to do exact arithmetic buried inside a reading-comprehension task, and small models in particular are not reliable at that combination. This is the first real lesson: a single prompt that mixes “read the document” with “do the math” gives you no way to check which part went wrong. Splitting them into separate stages, one of which is not a model call at all, is what the rest of this tutorial builds.

Step 4: Design the state that flows between stages

Create models.py. Every stage below reads and returns one of these plain dataclasses, so the pipeline’s shape is visible just from this one file:

from dataclasses import dataclass, field

@dataclass
class SLATier:
    below_pct: float
    at_or_above_pct: float
    credit_pct: float

@dataclass
class SLATerms:
    target_pct: float
    tiers: list[SLATier]
    excused_categories: list[str]
    raw_source: str = ""

@dataclass
class UptimeReport:
    total_minutes: int
    counted_downtime_minutes: int
    excused_downtime_minutes: int
    uptime_pct: float
    incident_count: int
    excused_incident_count: int

@dataclass
class ReconciliationResult:
    selected_credit_pct: float | None
    justification: str
    raw_response: str = ""

@dataclass
class ValidationIssue:
    stage: str
    message: str

@dataclass
class ValidationResult:
    passed: bool
    issues: list[ValidationIssue] = field(default_factory=list)
    expected_credit_pct: float | None = None

Nothing here calls a model. This file only defines the shape of the data as it moves from stage 1 (SLATerms) through stage 2 (UptimeReport) into stage 3 (ReconciliationResult) and finally stage 4 (ValidationResult). Keeping the state as typed dataclasses, rather than raw dictionaries, means a typo in a field name fails loudly instead of silently returning None somewhere downstream.

Step 5: Stage 1, extract structured terms from the contract

This stage’s job is to turn the contract’s prose into the SLATerms shape from Step 4. It is a genuine natural-language task, a regular expression cannot reliably pull “10% credit when uptime is below 99.9% but at or above 99.0%” out of a paragraph, so this is where a model earns its place in the pipeline. Create stage1_extract_terms.py:

import json
import ollama
from models import SLATerms, SLATier

MODEL_NAME = "qwen2.5:1.5b"

SCHEMA = {
    "type": "object",
    "properties": {
        "target_pct": {"type": "number"},
        "tiers": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "below_pct": {"type": "number"},
                    "at_or_above_pct": {"type": "number"},
                    "credit_pct": {"type": "number"},
                },
                "required": ["below_pct", "at_or_above_pct", "credit_pct"],
            },
        },
        "excused_categories": {"type": "array", "items": {"type": "string"}},
    },
    "required": ["target_pct", "tiers", "excused_categories"],
}

PROMPT = """You are extracting Service Level Agreement terms from a contract.
Read the contract text below and return ONLY the requested JSON.

Rules:
- "tiers" lists every service-credit tier the contract defines. Each tier has
  below_pct (the upper, exclusive uptime bound), at_or_above_pct (the lower,
  inclusive uptime bound), and credit_pct (the credit percentage owed).
- "excused_categories" lists the short category names for downtime the
  contract says should NOT count against uptime (for example, planned
  maintenance that was announced in advance). Use short snake_case labels
  like "scheduled_maintenance", not full sentences.
- "target_pct" is the uptime commitment percentage the contract states
  up front, before any credit schedule.

CONTRACT TEXT:
---
{contract_text}
---

Return only the JSON object, nothing else.
"""

def extract_terms(contract_text, model=MODEL_NAME):
    response = ollama.chat(
        model=model,
        messages=[{"role": "user", "content": PROMPT.format(contract_text=contract_text)}],
        format=SCHEMA,
        options={"temperature": 0},
    )
    raw = response["message"]["content"]
    data = json.loads(raw)
    tiers = [SLATier(**t) for t in data["tiers"]]
    return SLATerms(
        target_pct=data["target_pct"],
        tiers=tiers,
        excused_categories=data["excused_categories"],
        raw_source=raw,
    )

The format=SCHEMA argument is Ollama’s structured outputs feature, which its own announcement describes plainly: it “makes it possible to constrain a model’s output to a specific format defined by a JSON schema.” That constraint is real and useful, but pay close attention to what it does and does not promise. It guarantees the response will parse as valid JSON with the right field names and types (a number where a number is required, a list of strings where a list of strings is required). It says nothing at all about whether the values inside those fields are correct.

To see the difference, add a second, trickier contract. Create contract_b.txt, which states the exact same terms as contract_a.txt but written the way an actual services agreement is often written, as dense legalese with numbers spelled out in words as well as digits:

HARBORLIGHT DATA SYSTEMS: MASTER SERVICES AGREEMENT, SCHEDULE 4 (AVAILABILITY)

4.1 Availability Target. Subject to the exclusions described in Section 4.3,
Provider shall use commercially reasonable efforts to attain a Monthly
Availability Percentage of no less than ninety-nine point nine percent
(99.9%) in each calendar month during the Term.

4.2 Remedies for Shortfall. In the event Provider fails to attain the
Availability Target set out in Section 4.1 for a given calendar month, and
without limiting any other remedy available to Customer at law or in equity,
Customer shall, upon written request submitted within thirty (30) days of
the end of the affected month, be entitled to a credit against amounts
otherwise payable for that month, calculated as follows: where Monthly
Availability Percentage for the affected month is less than 99.9% but not
less than 99.0%, the credit shall equal ten percent (10%) of the fees
attributable to that month; where Monthly Availability Percentage is less
than 99.0% but not less than 95.0%, the credit shall equal twenty-five
percent (25%) of such fees; and where Monthly Availability Percentage falls
below 95.0%, the credit shall equal fifty percent (50%) of such fees, it
being understood that in no event shall aggregate credits under this
Schedule 4 exceed fifty percent (50%) of the fees payable for the month in
question.

4.3 Exclusions. Notwithstanding anything to the contrary herein, any period
of unavailability attributable to (a) Scheduled Maintenance for which
Provider has given Customer no less than twenty-four (24) hours' advance
notice, (b) Force Majeure Events, or (c) acts or omissions of Customer or
its End Users, shall be excluded from the calculation of Monthly
Availability Percentage and shall not be deemed a failure to attain the
Availability Target.

Now run stage 1 against both contracts:

# Context: venv activated, contract_a.txt, contract_b.txt, stage1_extract_terms.py all in the project folder.
# Purpose: extract structured tiers from each contract and print them.
python -c "
from stage1_extract_terms import extract_terms
for path in ['contract_a.txt', 'contract_b.txt']:
    with open(path, encoding='utf-8') as f:
        terms = extract_terms(f.read())
    print(f'=== {path} ===')
    print('target_pct:', terms.target_pct)
    print('excused_categories:', terms.excused_categories)
    for t in terms.tiers:
        print(f'  below={t.below_pct} at_or_above={t.at_or_above_pct} credit={t.credit_pct}')
"

Expected output (captured from an actual run, unedited):

=== contract_a.txt ===
target_pct: 99.9
excused_categories: ['scheduled_maintenance']
  below=0.1 at_or_above=0.0 credit=0.1
  below=0.25 at_or_above=0.0 credit=0.25
  below=0.5 at_or_above=0.0 credit=0.5

=== contract_b.txt ===
target_pct: 99.9
excused_categories: ['scheduled_maintenance']
  below=0.1 at_or_above=0.9 credit=50
  below=0.2 at_or_above=0.9 credit=50

Both extractions are wrong, in two different ways. For contract_a.txt, the model got target_pct and the excused category right, but every tier value looks like the percentage written as a fraction (10% read back as 0.1, 25% as 0.25) with the bounds scrambled on top: none of below=0.1, below=0.25, or below=0.5 corresponds to anything the contract actually says. For contract_b.txt, the model dropped an entire tier (only two came back instead of three) and produced bounds that are backwards: at_or_above=0.9 is larger than below=0.1, which describes a range that cannot contain any number at all. All of this is valid JSON. It matched the schema on every field. This is the gap between “structurally valid” and “correct,” and it is the same distinction this site’s earlier tutorial on structured LLM output covers for a single extraction call; here it matters even more, because stage 1’s output becomes stage 3’s input.

Step 6: Stage 2, compute uptime without a model

Given the incident log and the set of excused categories stage 1 extracted, computing uptime is exact arithmetic: total minutes in the month, minus non-excused downtime, divided by total minutes. Python does this precisely, every time, for free. Handing it to a language model would mean re-verifying its arithmetic on every single run for no benefit, which is exactly what Step 3’s one-shot comparison demonstrated going wrong. Create stage2_uptime.py:

import csv
from models import UptimeReport

MINUTES_PER_DAY = 24 * 60

def load_incidents(path):
    with open(path, newline="", encoding="utf-8") as f:
        return list(csv.DictReader(f))

def compute_uptime(incidents, days_in_month, excused_categories):
    total_minutes = days_in_month * MINUTES_PER_DAY
    counted = 0
    excused = 0
    excused_count = 0
    for row in incidents:
        minutes = int(row["minutes"])
        if row["category"] in excused_categories:
            excused += minutes
            excused_count += 1
        else:
            counted += minutes
    uptime_pct = (total_minutes - counted) / total_minutes * 100
    return UptimeReport(
        total_minutes=total_minutes,
        counted_downtime_minutes=counted,
        excused_downtime_minutes=excused,
        uptime_pct=round(uptime_pct, 4),
        incident_count=len(incidents),
        excused_incident_count=excused_count,
    )

Notice that this stage’s correctness genuinely depends on stage 1: if the extracted excused_categories list is wrong, empty, or missing the real category name, this deterministic calculation will confidently produce the wrong uptime figure even though its own math never makes a mistake. Run it both ways to see the size of that dependency:

python -c "
from stage2_uptime import load_incidents, compute_uptime
incidents = load_incidents('incidents_april_2026.csv')
print('Correct (excusing scheduled_maintenance):')
print(compute_uptime(incidents, 30, excused_categories=['scheduled_maintenance']))
print()
print('Naive (nothing excused):')
print(compute_uptime(incidents, 30, excused_categories=[]))
"

Expected output:

Correct (excusing scheduled_maintenance):
UptimeReport(total_minutes=43200, counted_downtime_minutes=150, excused_downtime_minutes=300, uptime_pct=99.6528, incident_count=5, excused_incident_count=2)

Naive (nothing excused):
UptimeReport(total_minutes=43200, counted_downtime_minutes=450, excused_downtime_minutes=0, uptime_pct=98.9583, incident_count=5, excused_incident_count=0)

99.6528% lands in the 10% credit tier. 98.9583%, just 0.69 percentage points lower, crosses a tier boundary and lands in the 25% tier. If stage 1 ever hands stage 2 an empty or wrong excused_categories list, the customer gets overcredited by 15 percentage points of their monthly fee, silently, with a perfectly correct-looking calculation behind it.

Step 7: Stage 3, decide the tier and justify it

The lookup itself (which tier does a given uptime percentage fall into) is simple enough to write as a Python one-liner. This stage still calls a model anyway, for a real reason: writing a clear, contract-referencing justification for a human reviewer to read is exactly the kind of prose task a language model is good at, and producing that sentence alongside the decision is part of what makes this an audit-ready pipeline rather than a bare number. Create stage3_reconcile.py:

import json
import ollama
from models import ReconciliationResult, SLATerms, UptimeReport

MODEL_NAME = "qwen2.5:1.5b"

SCHEMA = {
    "type": "object",
    "properties": {
        "selected_credit_pct": {"type": ["number", "null"]},
        "justification": {"type": "string"},
    },
    "required": ["selected_credit_pct", "justification"],
}

PROMPT = """You are reconciling a monthly SLA credit. Given the contract's
credit tiers and the measured Monthly Uptime Percentage for the month,
decide which tier (if any) applies and justify the decision in one or two
sentences, citing the specific tier bounds.

If the uptime is at or above the target percentage, no tier applies and
selected_credit_pct must be null.

CREDIT TIERS (JSON):
{tiers_json}

MEASURED MONTHLY UPTIME PERCENTAGE: {uptime_pct}%

Return only the JSON object, nothing else.
"""

def reconcile(terms, uptime, model=MODEL_NAME):
    tiers_json = json.dumps([
        {"below_pct": t.below_pct, "at_or_above_pct": t.at_or_above_pct, "credit_pct": t.credit_pct}
        for t in terms.tiers
    ])
    response = ollama.chat(
        model=model,
        messages=[{"role": "user", "content": PROMPT.format(
            tiers_json=tiers_json, uptime_pct=uptime.uptime_pct)}],
        format=SCHEMA,
        options={"temperature": 0},
    )
    raw = response["message"]["content"]
    data = json.loads(raw)
    return ReconciliationResult(
        selected_credit_pct=data["selected_credit_pct"],
        justification=data["justification"],
        raw_response=raw,
    )

To test this stage in isolation from stage 1’s problems, feed it correctly extracted, hand-verified terms and check its answer against every tier boundary:

from models import SLATerms, SLATier, UptimeReport
from stage3_reconcile import reconcile

CORRECT_TERMS = SLATerms(
    target_pct=99.9,
    tiers=[
        SLATier(below_pct=99.9, at_or_above_pct=99.0, credit_pct=10),
        SLATier(below_pct=99.0, at_or_above_pct=95.0, credit_pct=25),
        SLATier(below_pct=95.0, at_or_above_pct=0.0, credit_pct=50),
    ],
    excused_categories=["scheduled_maintenance"],
)

for pct in [99.95, 99.9, 99.65, 99.0, 98.99, 95.0, 94.99, 50.0]:
    uptime = UptimeReport(43200, 0, 0, pct, 0, 0)
    r = reconcile(CORRECT_TERMS, uptime)
    print(f"uptime={pct:>6}%  selected={r.selected_credit_pct}")

Real, unedited output from that loop is on the left below. The right-hand notes are not part of what the script prints, they are the deterministically correct tier for each uptime (10, 25, 50, or none once uptime meets the 99.9% target), added here so you can see at a glance which lines are right:

uptime= 99.95%  selected=None      # correct: no tier applies
uptime=  99.9%  selected=None      # correct: no tier applies
uptime= 99.65%  selected=None      # WRONG: should be 10
uptime=  99.0%  selected=None      # WRONG: should be 10
uptime= 98.99%  selected=None      # WRONG: should be 25
uptime=  95.0%  selected=None      # WRONG: should be 25
uptime= 94.99%  selected=None      # WRONG: should be 50
uptime=  50.0%  selected=None      # WRONG: should be 50

Even with completely correct, valid, unambiguous input, this stage returns None almost every time, including for 50.0% uptime, a case with obvious real downtime that plainly deserves the largest credit tier. Look at the raw justification the model gave for that last case, captured directly from a real response:

"The measured Monthly Uptime Percentage of 50.0% is below the 'below_pct'
threshold of 99.9%, so no credit tier applies."

Read that sentence closely and the bug becomes visible: the model has the direction backwards. Being below a tier’s below_pct is exactly when that tier should apply, not a reason to reject every tier. It appears to be checking only the first tier’s upper bound and treating “below the SLA’s headline commitment” as disqualifying, rather than checking both bounds of each tier in turn the way the deterministic lookup in Step 8 actually does. This is not a formatting problem structured output could have caught, since null is a perfectly valid response under the schema. It is a reasoning mistake wearing a well-typed JSON object.

Does a clearer prompt fix it?

It is worth trying to fix this with a better prompt before reaching for a safety net, if only to see how far that gets you. Here is a second attempt, stage3_reconcile_v2.py, that spells the comparison out as an explicit row-by-row lookup instead of leaving the model to infer the logic on its own:

PROMPT = """You are checking which row of a lookup table matches a number.

TABLE (each row is one tier, 0-indexed):
{rows}

NUMBER TO LOOK UP: {uptime}

For each row in order starting at index 0, apply this exact test:
  row.at_or_above_pct <= NUMBER  AND  NUMBER < row.below_pct
The first row where BOTH parts are true is the match. A smaller
at_or_above_pct means a WORSE (more severe) tier, not a safer one.
If NUMBER is greater than or equal to every row's below_pct, there is
no match: return null for both fields.

Return only JSON: matched_tier_index (the row index that matched, or
null), selected_credit_pct (that row's credit_pct, or null), and a one
sentence justification naming the row bounds you checked.
"""

Re-running the same eight boundary values against this clearer prompt:

uptime= 99.95%  got=25   expected=None  [MISMATCH]
uptime=  99.9%  got=25   expected=None  [MISMATCH]
uptime= 99.65%  got=25   expected=10    [MISMATCH]
uptime=  99.0%  got=10   expected=10    [OK]
uptime= 98.99%  got=25   expected=25    [OK]
uptime=  95.0%  got=25   expected=25    [OK]
uptime= 94.99%  got=25   expected=50    [MISMATCH]
uptime=  50.0%  got=25   expected=50    [MISMATCH]

Genuinely better: 3 of 8 correct instead of 2, and this time all three of those correct answers are real, non-null tier matches rather than the “no tier applies” default that accounted for both correct answers under the first prompt. Genuinely still broken: 5 of 8 wrong, and one of those wrong answers is the most dangerous kind, assigning a 25% credit to 99.95% uptime, which is above the target and should carry no credit at all. The model appears to default to row 1 whenever it is uncertain rather than correctly rejecting the lookup. Spending more effort on the prompt bought a real, measurable improvement and did not come close to making the stage trustworthy on its own. That is the honest result, and it is the reason Step 8 exists: no amount of prompt tuning replaces an independent check, because you cannot know in advance which of your carefully worded instructions the model will follow this time.

Step 8: Stage 4, the validation gate

This stage never calls a model. It runs two independent checks that do not trust anything upstream: a structural sanity check on stage 1’s own extracted tiers, and a deterministic cross-check that recomputes which tier the uptime should fall into and compares that against whatever stage 3 selected. Any mismatch fails the whole pipeline closed instead of letting a number through. Write a first version in stage4_validate.py:

from models import SLATerms, UptimeReport, ReconciliationResult, ValidationResult, ValidationIssue

def _expected_tier_credit(terms, uptime_pct):
    if uptime_pct >= terms.target_pct:
        return None
    for tier in sorted(terms.tiers, key=lambda t: t.at_or_above_pct, reverse=True):
        if tier.at_or_above_pct <= uptime_pct < tier.below_pct:
            return tier.credit_pct
    return None

def validate(terms, uptime, reconciliation):
    issues = []

    if not (1 <= len(terms.tiers) <= 6):
        issues.append(ValidationIssue("terms", f"unlikely tier count: {len(terms.tiers)}"))

    for t in terms.tiers:
        if not (0 <= t.below_pct <= 100):
            issues.append(ValidationIssue("terms", f"below_pct out of range: {t.below_pct}"))
        if not (0 <= t.at_or_above_pct <= 100):
            issues.append(ValidationIssue("terms", f"at_or_above_pct out of range: {t.at_or_above_pct}"))
        if not (0 <= t.credit_pct <= 100):
            issues.append(ValidationIssue("terms", f"credit_pct out of range: {t.credit_pct}"))
        if t.at_or_above_pct >= t.below_pct:
            issues.append(ValidationIssue(
                "terms", f"tier bounds inverted: at_or_above={t.at_or_above_pct} >= below={t.below_pct}"))

    if issues:
        return ValidationResult(passed=False, issues=issues, expected_credit_pct=None)

    expected = _expected_tier_credit(terms, uptime.uptime_pct)
    got = reconciliation.selected_credit_pct
    if expected != got:
        issues.append(ValidationIssue(
            "reconciliation",
            f"stage 3 selected {got!r} but independent recomputation expects {expected!r} "
            f"for uptime {uptime.uptime_pct}%"))
        return ValidationResult(passed=False, issues=issues, expected_credit_pct=expected)

    return ValidationResult(passed=True, issues=[], expected_credit_pct=expected)

Before trusting this, write tests against it, including a test built directly from the real broken output stage 1 produced in Step 5:

def test_validate_flags_fraction_instead_of_percentage():
    broken_terms = SLATerms(
        target_pct=99.9,
        tiers=[SLATier(below_pct=0.1, at_or_above_pct=0.0, credit_pct=0.1)],
        excused_categories=["scheduled_maintenance"],
    )
    uptime = UptimeReport(43200, 150, 300, 99.6528, 5, 2)
    reconciliation = ReconciliationResult(selected_credit_pct=None, justification="n/a")
    result = validate(broken_terms, uptime, reconciliation)
    assert not result.passed

Run it:

python -m pytest test_pipeline.py -v -k fraction

Expected output (the real result, not what you would want to see):

FAILED test_pipeline.py::test_validate_flags_fraction_instead_of_percentage
AssertionError: assert not True

The test failed. validate() returned passed=True for input that is obviously wrong. The reason is visible if you check each range condition by hand: below_pct=0.1, at_or_above_pct=0.0, and credit_pct=0.1 all satisfy 0 <= x <= 100, and 0.0 is not greater than or equal to 0.1, so the inverted-bounds check does not fire either. Every check in the first version genuinely passes on genuinely wrong data, because “is this number between 0 and 100” is a much weaker question than “does this number make sense for an SLA.” A plain range check is not the same thing as a domain sanity check, and the gap between them is exactly where this real bug lived.

The fix is a business-rule check specific to this domain: the tier table’s own highest bound should sit right where the target commitment stops being met. Add this to validate(), right before the final if issues: line that was already there:

    # A tier table only makes sense relative to the target: the first tier
    # (the mildest shortfall) should start right where the commitment stops
    # being met. Values like 0.1 or 0.0 pass the plain 0-100 range check
    # above but are nowhere near a target_pct of 99.9, so this catches the
    # fraction-vs-percentage confusion the range check alone misses.
    if terms.tiers and not issues:
        highest_tier_bound = max(t.below_pct for t in terms.tiers)
        if abs(highest_tier_bound - terms.target_pct) > 1.0:
            issues.append(ValidationIssue(
                "terms",
                f"top tier bound {highest_tier_bound} is not close to target_pct "
                f"{terms.target_pct}; extraction likely used the wrong units"))

Run the full suite again:

python -m pytest test_pipeline.py -v

Expected output:

test_pipeline.py::test_uptime_excludes_scheduled_maintenance PASSED
test_pipeline.py::test_uptime_naive_counts_everything PASSED
test_pipeline.py::test_validate_passes_when_reconciliation_matches PASSED
test_pipeline.py::test_validate_flags_wrong_tier_selection PASSED
test_pipeline.py::test_validate_flags_fraction_instead_of_percentage PASSED
test_pipeline.py::test_validate_flags_inverted_tier_bounds PASSED
test_pipeline.py::test_validate_returns_none_when_uptime_meets_target PASSED
test_pipeline.py::test_boundary_uptime_is_inclusive_of_the_lower_bound PASSED

============================== 8 passed in 0.02s ==============================

All eight pass, run entirely without a model, in about 20 milliseconds. None of them call Ollama: stage 1 and stage 3 outputs are constructed directly in the test file, so the suite is fast and never depends on what the model happens to say on a given day.

Step 9: Wire it all together

Create pipeline.py to chain all four stages and print a report:

from stage1_extract_terms import extract_terms
from stage2_uptime import compute_uptime, load_incidents
from stage3_reconcile import reconcile
from stage4_validate import validate

def run_pipeline(contract_path, incidents_path, days_in_month, model="qwen2.5:1.5b"):
    with open(contract_path, encoding="utf-8") as f:
        contract_text = f.read()

    terms = extract_terms(contract_text, model=model)
    incidents = load_incidents(incidents_path)
    uptime = compute_uptime(incidents, days_in_month, excused_categories=terms.excused_categories)
    reconciliation = reconcile(terms, uptime, model=model)
    validation = validate(terms, uptime, reconciliation)

    print(f"--- {contract_path} / {model} ---")
    print(f"Extracted tiers: {terms.tiers}")
    print(f"Uptime: {uptime.uptime_pct}%")
    print(f"Stage 3 selected credit: {reconciliation.selected_credit_pct}")
    print(f"Validation passed: {validation.passed}")
    if not validation.passed:
        for issue in validation.issues:
            print(f"  ISSUE [{issue.stage}]: {issue.message}")
        print(f"  Independently expected credit: {validation.expected_credit_pct}")
    return terms, uptime, reconciliation, validation

if __name__ == "__main__":
    run_pipeline("contract_a.txt", "incidents_april_2026.csv", 30)
    run_pipeline("contract_b.txt", "incidents_april_2026.csv", 30)

# Context: venv activated, all six project files in the same folder.
# Purpose: run the complete four-stage pipeline end to end on both contracts.
python pipeline.py

Expected output (captured from an actual run):

--- contract_a.txt / qwen2.5:1.5b ---
Extracted tiers: [SLATier(below_pct=0.1, at_or_above_pct=0.0, credit_pct=0.1), SLATier(below_pct=0.25, at_or_above_pct=0.0, credit_pct=0.25), SLATier(below_pct=0.5, at_or_above_pct=0.0, credit_pct=0.5)]
Uptime: 99.6528%
Stage 3 selected credit: None
Validation passed: False
  ISSUE [terms]: top tier bound 0.5 is not close to target_pct 99.9; extraction likely used the wrong units
  Independently expected credit: None

--- contract_b.txt / qwen2.5:1.5b ---
Extracted tiers: [SLATier(below_pct=0.1, at_or_above_pct=0.9, credit_pct=50), SLATier(below_pct=0.2, at_or_above_pct=0.9, credit_pct=50)]
Uptime: 99.6528%
Stage 3 selected credit: None
Validation passed: False
  ISSUE [terms]: tier bounds inverted: at_or_above=0.9 >= below=0.1
  ISSUE [terms]: tier bounds inverted: at_or_above=0.9 >= below=0.2
  Independently expected credit: None

This is the whole tutorial’s point, watched happening end to end. For both contracts, stage 1’s extraction is genuinely broken, stage 3 reaches a confident, well-punctuated, completely wrong conclusion (“no tier applies”) built on top of that broken extraction, and stage 4 catches it before it reaches an invoice. Without stage 4, this pipeline would have silently told a real customer they are owed nothing, on a month where they are actually owed a 10% credit. With it, the pipeline correctly refuses to answer and reports exactly which upstream stage to go check, by name, instead of returning a confident, wrong number.

Common mistakes and gotchas

• Assuming schema-constrained output means correct output. Every one of stage 1’s broken extractions in this tutorial was valid, schema-conforming JSON. Structured output guarantees shape, not truth; you still need a semantic check on top, which is what stage 4’s units-sanity rule is for.
• Writing range checks and calling it validation. This tutorial’s own first validator passed genuinely broken data because 0.0, 0.1, and 0.1 are all technically between 0 and 100. A useful validation rule usually has to know something specific about the domain, here, that a tier table’s bounds should sit near the stated target percentage, not just that its numbers fall in a generic range.
• Assuming a bigger model fixes a reasoning bug. This tutorial deliberately used a small, 1.5-billion-parameter model so its mistakes would be frequent and easy to reproduce, but do not assume switching to a larger one is a free win. A repeat attempt at the same extraction with the 4-billion-parameter qwen3.5:4b model returned an empty response and crashed the JSON parser on one run, and simply hung past 90 seconds with no output at all on another, almost certainly the same default “thinking” behavior documented in an earlier tutorial on this site that consumed its entire response budget on hidden reasoning. A bigger model can trade one failure mode for a different, harder-to-debug one; the validation gate has to catch both.
• Treating stages as independent when their correctness is coupled. Stage 2’s arithmetic is exact, but it is only as correct as the excused_categories list stage 1 handed it. A pipeline stage that never touches a model can still be wrong, if what feeds it is wrong.

How to confirm it all works end to end

1. Run python -m pytest test_pipeline.py -v and confirm all 8 tests pass in well under a second. These never touch Ollama, so a failure here means a logic bug in your own code, not a flaky model response.
2. Run python pipeline.py and confirm Validation passed: False appears for both contracts, each with an ISSUE line naming the actual problem in stage 1’s extraction. If you ever see Validation passed: True with a nonzero selected_credit_pct on this exact dataset, treat that as suspicious rather than as a fix, since Step 7 showed the current prompt has no verified case where it is correct on a nonzero tier without a model upgrade or a prompt rewrite you have separately tested.
3. Run python one_shot_comparison.py once more and confirm it still returns a 96.5% uptime figure that does not match the 99.6528% you can compute by hand from incidents_april_2026.csv. This confirms the one-shot failure mode is repeatable, not a one-off fluke.

Next steps

From here, a natural extension is to make the validation gate itself the thing that decides whether to retry: instead of just failing closed, have it re-run stage 1 with a stricter prompt, or fall back to a human review queue, the same escalation pattern the Red Hat quickstart that inspired this tutorial uses for its own variance-detection stage. If you want to see the single-agent version of “check your own work,” this site’s LangGraph tutorial builds a generate-critique-retry loop that solves a related but different problem: one agent looping back on itself, rather than several fixed stages handing off to each other. And if you want to push further on stage 3’s reconciliation logic specifically, try replacing the LLM call there with a deterministic lookup entirely, then compare how much of this pipeline’s value actually came from the model versus from validating it; the honest answer, based on this tutorial’s own numbers, is that stage 1’s extraction is the one step here a plain script genuinely could not have replaced.