Designing Large-Volume AI Processing Pipelines

Once you've decided a workload belongs in batch (Lesson 1), the next problem is architectural, not mechanical. Unit 12, Lesson 5 showed you how to build a JSONL file, submit it to the Batch API, and read back the output file. That is the submission mechanism. It says nothing about how thousands or millions of source records get turned into well-formed requests in the first place, how results get matched back to your database, or how the whole thing survives a bug in item #40,000 without corrupting the other 39,999. That is pipeline design, and it's the actual engineering work behind any production AI processing system.

What a Pipeline Actually Is

A pipeline is a sequence of independent stages that data flows through, where each stage has a single, well-defined responsibility and communicates with the next stage through a stable, inspectable format — typically records in a database table, rows in a file, or messages in a queue. The alternative — one large function that reads source data, builds prompts, calls the model, parses results, and writes to the destination all in one pass — works for a quick script but becomes unmaintainable and unrecoverable at scale. If that single function crashes on item 40,000, you often cannot tell what succeeded, and you have no seam at which to insert monitoring, retries, or a human review step.

A typical large-volume AI processing pipeline has five stages:

Ingest → Prepare → Submit → Collect → Finalize

Ingest: Pull the raw source records from wherever they live — a database table, a CSV export, an object storage bucket — and normalize them into a consistent internal representation. This stage should not know anything about prompts or models.

Prepare: Turn each normalized record into a model request: build the prompt, attach the correct model parameter, set response_format or any structured output schema, and assign a stable, unique identifier to the request. This is where prompt templates live, and it should not know anything about how the request is submitted.

Submit: Hand off the prepared requests to whichever execution mechanism you're using — the Batch API's JSONL upload, or a pool of concurrent async requests (Lessons 3-5 of this unit). This stage is a thin adapter and should be swappable without touching Prepare or Finalize.

Collect: Gather results as they become available, matching each result back to its original record using the identifier assigned in Prepare.

Finalize: Parse and validate each result, write it to its permanent destination, and record success or failure per item so the job's status is queryable afterward.

Why This Separation Matters

The value of splitting a pipeline into stages isn't aesthetic — it directly determines what happens when things go wrong, which at scale is not a matter of if but when. Consider what a stage boundary buys you:

• Independent failure isolation. If Submit fails halfway through (a network outage, a rate-limit exhaustion), the records that already passed through Prepare are untouched and don't need to be rebuilt. You resume from Submit, not from Ingest.
• Independent scaling and technology choices. Ingest might run once as a batch SQL query. Submit might run as an async worker pool. These can use completely different execution models because they only communicate through a shared record format.
• Testability. Each stage can be unit-tested with fake inputs and no network calls, because its contract is "takes records in shape A, returns records in shape B."
• Observability. You can count how many records are sitting in each stage at any moment, which immediately tells you where a stuck pipeline is stuck.

Designing the Record: The Central Data Structure

Every stage passes the same conceptual unit forward: one record, evolving as it moves through the pipeline. Defining this shape explicitly, rather than passing around loose dictionaries, prevents an enormous class of bugs where one stage silently expects a field another stage forgot to set.

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional


class RecordStatus(str, Enum):
    PENDING = "pending"
    PREPARED = "prepared"
    SUBMITTED = "submitted"
    SUCCEEDED = "succeeded"
    FAILED = "failed"


@dataclass
class PipelineRecord:
    record_id: str              # stable identifier, e.g. a DB primary key
    source_data: dict           # original input fields
    status: RecordStatus = RecordStatus.PENDING
    prompt: Optional[str] = None
    model_response: Optional[str] = None
    error: Optional[str] = None
    attempts: int = 0
    metadata: dict = field(default_factory=dict)


def prepare_record(record: PipelineRecord, prompt_template: str) -> PipelineRecord:
    """The Prepare stage: builds the prompt, does not call the model."""
    record.prompt = prompt_template.format(**record.source_data)
    record.status = RecordStatus.PREPARED
    return record

The record_id field deserves special attention: it must be stable and unique, and it must be something you can regenerate deterministically from the source data (a database row's primary key, not a randomly generated value created fresh on every pipeline run). This identifier is what lets the Collect stage match a model response back to the correct record, and it's the foundation that Lesson 9's resumability design depends on — without a stable ID, you cannot tell whether "record 40,000 in this run" is the same logical record as "record 40,000 in the retry."

A Minimal End-to-End Skeleton

The following sketch wires the five stages together using plain function calls (later lessons replace Submit and Collect with real Batch API or async concurrency code). The point of this example is the shape of the pipeline, not the execution mechanism.

from typing import Callable


def run_pipeline(
    source_records: list[dict],
    prompt_template: str,
    submit_and_collect: Callable[[list[PipelineRecord]], list[PipelineRecord]],
) -> list[PipelineRecord]:
    # Ingest
    records = [
        PipelineRecord(record_id=str(row["id"]), source_data=row)
        for row in source_records
    ]

    # Prepare
    records = [prepare_record(r, prompt_template) for r in records]

    # Submit + Collect (mechanism-specific, injected as a function)
    records = submit_and_collect(records)

    # Finalize
    for r in records:
        if r.status == RecordStatus.SUCCEEDED:
            save_result(r)
        else:
            log_failure(r)

    return records


def save_result(record: PipelineRecord) -> None:
    print(f"Saved result for {record.record_id}: {record.model_response[:40]!r}")


def log_failure(record: PipelineRecord) -> None:
    print(f"Failed {record.record_id} after {record.attempts} attempts: {record.error}")

Notice that run_pipeline takes submit_and_collect as a parameter rather than hardcoding a call to the OpenAI client. This is a form of dependency injection: it lets you test the pipeline's ingest/prepare/finalize logic with a fake submission function that returns canned results instantly, with no network access and no cost, which is exactly the testing pattern used throughout this unit.

def fake_submit_and_collect(records: list[PipelineRecord]) -> list[PipelineRecord]:
    for r in records:
        r.status = RecordStatus.SUCCEEDED
        r.model_response = f"summary of {r.source_data['id']}"
    return records


def test_pipeline_runs_end_to_end():
    source = [{"id": 1, "text": "hello"}, {"id": 2, "text": "world"}]
    results = run_pipeline(source, "Summarize: {text}", fake_submit_and_collect)
    assert all(r.status == RecordStatus.SUCCEEDED for r in results)
    assert results[0].model_response == "summary of 1"
    print("PASS: pipeline runs end to end with fake submission")


test_pipeline_runs_end_to_end()

This test never touches the network and runs in milliseconds, yet it verifies the entire wiring of the pipeline — exactly the kind of test you want in a CI suite that runs on every commit, distinct from a separate, much smaller set of manual smoke tests that exercise the real Batch API.

Where State Lives

A large-volume pipeline processing more items than fit comfortably in memory, or one that must survive a process restart, needs its record state persisted somewhere durable — a database table with one row per record_id and a status column is the most common choice. Lesson 6 (progress tracking) and Lesson 9 (resumability) both build directly on this idea: if PipelineRecord.status is only ever held in a Python list in memory, a crash loses all progress information and the entire job must restart from scratch. Deciding now, at design time, that status belongs in a durable store — not because you need resumability yet, but because retrofitting persistence into a pipeline that was designed to be memory-only is far more disruptive than including it from the start.

Common Mistakes

• Collapsing all five stages into one function "for simplicity." This is the single most common reason batch pipelines become unmaintainable. It feels faster to write initially, but it means a single exception anywhere aborts the entire run with no way to know which records had already succeeded.
• Using array position instead of a stable ID to match requests to responses. If any stage filters, reorders, or retries a subset of records, positional matching silently associates the wrong response with the wrong record. Always carry an explicit record_id.
• Mixing prompt-building logic into the submission stage. This makes it impossible to swap the Batch API for async concurrent calls (or vice versa) without duplicating your prompt logic in two places, and it makes prompt logic hard to unit test in isolation.

Best Practices

• Define the record schema before writing any stage. Treat PipelineRecord (or your equivalent) as a contract the whole pipeline agrees on, and version it deliberately if it needs to change once records exist in a persisted store.
• Make Submit and Collect swappable. Inject them as parameters or behind a small interface, exactly as submit_and_collect was injected above, so the same Ingest/Prepare/Finalize logic works whether the execution mechanism is the Batch API, async concurrency, or a future replacement.
• Persist status transitions, not just final results. Knowing that a record is currently SUBMITTED versus PREPARED versus FAILED is what makes a stuck or crashed pipeline diagnosable, and it's the foundation the rest of this unit builds on.