Document pipeline

I built a document pipeline for an Australian state government client. It reads regulatory compliance documents, extracts each table row on its own, checks it, and links every value back to the page and cell it came from. The records then load into the client's CRM. The measured results are under Results.

The problem

The documents are long PDFs full of tables. Rows carry hierarchical records: a parent row gives context to the child rows beneath it. Every value has to end up in Salesforce custom objects, and it has to pass the CRM's validation rules before it will import.

Until now that meant manual data entry. It was slow, transcription errors crept in, and reporting further down the line waited on it.

A few things made automation hard:

• Layouts vary. Different authors order columns differently and use different table structures and headings.
• Tables span pages. Headers repeat mid-document and rows split across page breaks.
• Records nest. A child row means nothing without its parent, so the hierarchy has to survive extraction.
• Validation is strict. Some fields only accept values that depend on another field's value. One mismatch blocks the import.
• Every value needs a source. Reviewers must be able to see the exact page and table cell behind each extracted value.

How it works

The current pipeline is a 12-node LangGraph graph. A deterministic parser reads every table before any model call, and two human review gates sit between the graph and the CRM.

Each node checkpoints its output, so a failed run restarts from the last good step instead of from the top. Every attempt is traced, with tokens and cost per call, so a slow or costly step shows up.

A FastAPI backend streams progress to a Next.js review screen. A person checks flagged rows, opens the source page and approves the records at two gates before anything is exported.

The output is a set of structured records like the register below. The data is synthetic.

Decisions

import asyncio
import json
from collections.abc import AsyncGenerator
from dataclasses import dataclass, field
from typing import Any, Literal

from fastapi import APIRouter
from fastapi.responses import StreamingResponse

@dataclass
class PipelineEvent:
    event_type: Literal["stage_start", "record_extracted", "stage_complete", "error"]
    job_id: str
    payload: dict[str, Any] = field(default_factory=dict)

class PipelineEventBus:
    """Pipeline nodes publish; the SSE endpoint subscribes per job."""

    def __init__(self) -> None:
        self._queues: dict[str, asyncio.Queue[PipelineEvent | None]] = {}

    def register(self, job_id: str) -> asyncio.Queue[PipelineEvent | None]:
        queue: asyncio.Queue[PipelineEvent | None] = asyncio.Queue()
        self._queues[job_id] = queue
        return queue

    async def publish(self, event: PipelineEvent) -> None:
        queue = self._queues.get(event.job_id)
        if queue:
            await queue.put(event)

    async def close(self, job_id: str) -> None:
        queue = self._queues.pop(job_id, None)
        if queue:
            await queue.put(None)  # sentinel: end of stream

event_bus = PipelineEventBus()
router = APIRouter()

@router.get("/api/jobs/{job_id}/stream")
async def stream_job_events(job_id: str) -> StreamingResponse:
    queue = event_bus.register(job_id)

    async def generate() -> AsyncGenerator[str, None]:
        while (event := await queue.get()) is not None:
            yield f"data: {json.dumps(event.__dict__)}\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

Results

Read the F1 score as a benchmark result, not a general accuracy rate: larger document sets varied materially. The 190-page run is one observation and does not describe every document.

Status

I built it for an Australian state government client via Silvatron Pty Ltd, from 2025 to May 2026. It went into production as a managed service, with Langfuse for pipeline traces and Logfire for inspecting Pydantic validation.