Chaining the pipeline

By the end of this stage the document flow is a real pipeline. A runner executes ordered steps against a shared context, extraction routes to the right prebuilt model, and Northstar's finance policy runs as plain Python that produces typed issues.

The chaining pattern

With classification working, the question becomes how steps compose. Dave's prompt to the agent is deliberate about the shape, asking for a chain where "you can also easily add things, swap things, remove things" without breaking the rest. Later in the stage he makes the meta-point explicit: "design patterns can be really helpful, especially with AI agents." Asking for a named pattern got clean, extensible code instead of a tangle.

The result is three small pieces in backend/app/pipeline/base.py:

class PipelineContext(BaseModel):
    """Shared state passed through each pipeline step."""

    document_path: Path
    classification: DocumentClassification | None = None
    extraction: InvoiceExtraction | ReceiptExtraction | None = None
    review_data: ReviewData | None = None
    issues: list[ValidationIssue] | None = None
    gl_suggestion: GlSuggestion | None = None


class PipelineStep(Protocol):
    name: str

    def run(self, ctx: PipelineContext) -> PipelineContext: ...


class Pipeline:
    def __init__(self, steps: Sequence[PipelineStep]) -> None:
        self.steps = list(steps)

    def run(self, document_path: Path) -> PipelineContext:
        ctx = PipelineContext(document_path=document_path)
        for index, step in enumerate(self.steps, start=1):
            logger.info("[%d/%d] Starting step: %s", index, len(self.steps), step.name)
            ctx = step.run(ctx)
        return ctx

That is the whole engine, a loop over steps, each reading the context and returning an updated copy. Any object with a name and a run method is a step, which is what makes adding the GL step in the next stage a non-event. The context fields start as None and fill up as the document moves through, so each step can check that its prerequisites ran.

The finished app adds one more step between extraction and validation, the hybrid LLM review. It arrives in the Closing the loop stage; the chain absorbs it without changing anything here.

The extraction step

Extraction now becomes routing. The step reads the classification and picks the model:

class ExtractionStep:
    name = "extraction"

    def run(self, ctx: PipelineContext) -> PipelineContext:
        if ctx.classification is None:
            raise ValueError("ExtractionStep requires ctx.classification from ClassificationStep.")

        if ctx.classification.document_kind == DocumentKind.invoice:
            result = self._service.analyze_invoice(ctx.document_path)
            extraction = map_invoice_result(self._service.to_dict(result))
        else:
            result = self._service.analyze_receipt(ctx.document_path)
            extraction = map_receipt_result(self._service.to_dict(result))

        return ctx.model_copy(update={"extraction": extraction})

Everything it calls already exists, the service from the extraction stage and the mapping from the data-models stage. The pipeline is where the earlier building blocks click together.

Validation is code, not a model

This is the architectural rule from the brief made real: models extract evidence, code decides policy. The rules live as pure functions with no Azure calls anywhere near them:

from stdnum.eu import vat

if not data.vendor_vat_id:
    issues.append(
        _issue("vendor_vat_id_required", "vendor_vat_id", "Supplier VAT number is required.")
    )
elif not vat.is_valid(data.vendor_vat_id):
    issues.append(
        _issue(
            "vendor_vat_id_invalid",
            "vendor_vat_id",
            "Supplier VAT number has an invalid EU format or checksum.",
        )
    )

if (
    data.subtotal is not None
    and data.total_tax is not None
    and data.invoice_total is not None
    and abs(data.subtotal + data.total_tax - data.invoice_total) > Decimal("0.01")
):
    issues.append(
        _issue(
            "invoice_total_mismatch",
            "invoice_total",
            "Subtotal plus VAT does not match the invoice total.",
        )
    )

The full rule set for invoices requires a supplier name, a VAT number that passes the EU checksum via python-stdnum, a customer name and VAT matching Northstar's configured identity, an invoice number, date, total, and currency, a due date no earlier than the invoice date, a subtotal plus VAT that equals the total within one cent, and no duplicate from the same supplier, with a warning when the purchase order reference is missing. On top of that, any core field whose extraction confidence falls below 0.80 raises a low_confidence warning. Receipts get a shorter list, because as Dave notes, receipts typically carry no VAT number at all.

Every issue is a typed ValidationIssue with a stable code, a field, and a severity. Errors force review; warnings inform it. status_for_issues collapses the list into the document status. Any error means needs_review, otherwise ready.

Run the whole chain

cd playground
uv run --project ../backend --locked --no-sync python run_pipeline.py
# 14:03:12 INFO [app.pipeline.base] [1/3] Starting step: classification
# 14:03:15 INFO [app.pipeline.base] [2/3] Starting step: extraction
# 14:03:22 INFO [app.pipeline.base] [3/3] Starting step: validation

The script prints one payload with everything, the classification and its reasoning, an extraction summary, and the issue list. Run it against the happy path first, samples/generated/01-en-happy-classic.pdf, a clean invoice with a valid French VAT number, and the issue list comes back empty.

Then feed it trouble. Document 05 has no supplier VAT number, and the pipeline says so in a typed issue instead of a stack trace. Dave's reaction is the goal of the whole stage: "this is good, this is now what we can show to Maya," a pop-up that says exactly what is wrong with the document instead of silently accepting it.

Checkpoint

• The pipeline logs its numbered steps and completes on a sample document
• The happy-path invoice produces zero issues; document 05 produces vendor_vat_id_required
• You can explain why the VAT checksum and the totals check are code and not a prompt