Batch LLM Enrichment: Idempotency, Cost Per Row, and Knowing When the Output Is Wrong

The ticket said “classify support tickets by root cause.” A one-line prompt, an API key, a for loop over 40,000 rows. I had a working version in an hour and it felt like the easiest win of the quarter.

Then the DAG got retried after an unrelated upstream failure and re-billed the entire backfill. Then a JSON parse error at row 31,000 killed the task and the retry started from zero. Then someone changed one word in the prompt and three months of previously-classified tickets silently disagreed with the new ones, which nobody noticed until a QBR deck showed a category that had “grown 400%.”

None of those are AI problems. They are the problems of any expensive, non-deterministic, externally-hosted transformation, and data engineering has known how to handle expensive external calls for twenty years. The mistake is treating the LLM call as special instead of treating it as a step in a pipeline with the usual obligations: idempotency, incrementality, bounded cost, retry semantics, and tests.

This post is the set of patterns that survived. The strategic version of when to use an LLM at all is in LLMs in data pipelines; this is the implementation.

Model the enrichment as a table, not a step

The single decision that fixes most of this: the enrichment output lives in its own table, keyed by content hash and prompt version, never overwritten in place.

create table if not exists analytics.enrich.ticket_root_cause (
    content_hash   string      not null,   -- sha256 of the exact input text
    prompt_version string      not null,   -- bump on any prompt change
    model_id       string      not null,   -- e.g. 'gpt-5.2-2026-03'
    root_cause     string,
    confidence     float,
    input_tokens   int,
    output_tokens  int,
    cost_usd       float,
    latency_ms     int,
    status         string      not null,   -- ok | refused | invalid | error
    raw_response   variant,                -- keep it. you will need it.
    created_at     timestamp_ntz default current_timestamp(),
    primary key (content_hash, prompt_version, model_id)
);

The source table joins to this on the hash. Rows whose text has not changed and whose prompt version has not changed are already answered, so the incremental model selects only the misses:

-- models/enrich/tickets_to_enrich.sql
{{ config(materialized='view') }}

select
    t.ticket_id,
    t.body_text,
    sha2(t.body_text, 256) as content_hash
from {{ ref('stg_tickets') }} t
left join {{ source('enrich', 'ticket_root_cause') }} e
    on e.content_hash = sha2(t.body_text, 256)
   and e.prompt_version = '{{ var("prompt_version") }}'
   and e.model_id = '{{ var("model_id") }}'
   and e.status in ('ok', 'refused')   -- retry 'invalid' and 'error'
where e.content_hash is null
  and length(t.body_text) between 20 and 20000

This gives you three properties for free. Re-running the DAG costs nothing, because every row is already present. A prompt change becomes a controlled, priced backfill rather than a silent redefinition, because the old rows keep their old version and you can compare them. And a partial failure resumes exactly where it stopped, because success is recorded per row rather than per task.

That length filter matters more than it looks. In my ticket corpus about 4% of rows were empty, a single emoji, or a 90,000-character pasted log dump. Filtering them out before the API call removed 4% of cost and roughly 60% of my error rate.

flowchart TD
  src["Source rows"] --> hash["Content hash"]
  hash --> anti["Anti-join to enrichment table"]
  anti -->|"already answered"| skip["Skip, zero cost"]
  anti -->|"miss"| batch["Batch API submit"]
  batch --> parse{"Valid schema?"}
  parse -->|no| retryq["Retry up to 2, then status=invalid"]
  parse -->|yes| write["Write row with cost and tokens"]
  write --> qa["Sampled quality scoring"]

Idempotency is the anti-join. Everything downstream of it is ordinary batch engineering.

Set the cost per row before you write the prompt

Decide the budget first, then design backward. This is the step teams skip, and it is why LLM enrichment projects get cancelled after the second invoice.

The arithmetic is simple and worth doing on paper. My ticket classifier: roughly 600 input tokens (system prompt plus ticket body) and 40 output tokens per row. At a mid-tier model’s rates that is about $0.0009 per row. 40,000 tickets in the backfill is $36. Steady state at 1,200 new tickets a day is about $1.10 a day, or $33 a month. That is a number I can defend.

Now run the same arithmetic with a frontier model at roughly 10x the rate: $360 backfill, $330 a month. Still not enormous, but no longer obviously worth it for a categorical label — and this is exactly the comparison that should happen before anyone writes code.

Four levers, in the order I pull them:

Use the batch endpoint. Every major provider offers roughly 50% off for asynchronous batch submission with a 24-hour completion window. Batch enrichment in a nightly DAG has that window by definition. I default to batch and only use the synchronous API when something genuinely needs to be fresh within the hour.

Shrink the system prompt. It is charged on every single row. Mine started at 1,400 tokens with six few-shot examples. Cutting to three examples and tightening the instructions got it to 550 tokens with no measurable accuracy loss on the eval set — a 40% cost reduction from editing text.

Cap the input. Truncate to the first N characters of the ticket body rather than sending log dumps. I found no accuracy difference between 2,000 and 20,000 characters for root-cause classification, and a large difference in the bill.

Downgrade the model and prove it. Run your eval set against the cheap model. For structured classification with a good prompt, the gap is usually small. For summarization and extraction with nuance, it is not. Measure, do not assume.

Retries: distinguish the three failure kinds

A single retry=3 around the API call is wrong, because three different things fail and they need different handling.

import hashlib, json, time
from dataclasses import dataclass

import httpx
from pydantic import BaseModel, ValidationError, Field


class RootCause(BaseModel):
    root_cause: str = Field(pattern=r'^(billing|auth|performance|bug|howto|other)$')
    confidence: float = Field(ge=0.0, le=1.0)


@dataclass(frozen=True)
class Result:
    status: str          # ok | invalid | error | refused
    payload: dict | None
    cost_usd: float
    attempts: int


def enrich_one(client: httpx.Client, text: str, prompt: str) -> Result:
    cost = 0.0
    for attempt in range(1, 4):
        try:
            resp = client.post("/v1/responses", json={
                "model": MODEL_ID,
                "input": [{"role": "system", "content": prompt},
                          {"role": "user", "content": text[:8000]}],
                "response_format": {"type": "json_schema",
                                    "json_schema": RootCause.model_json_schema()},
                "temperature": 0,          # not determinism, but less variance
                "max_output_tokens": 80,
            }, timeout=60.0)
        except (httpx.TimeoutException, httpx.TransportError):
            time.sleep(2 ** attempt)       # transport: retry, no charge incurred
            continue

        if resp.status_code in (429, 500, 502, 503):
            # Respect the provider's backoff hint when it gives one.
            time.sleep(float(resp.headers.get("retry-after", 2 ** attempt)))
            continue
        if resp.status_code >= 400:
            return Result("error", {"http": resp.status_code}, cost, attempt)

        body = resp.json()
        cost += price(body["usage"])       # charged even if we reject the content

        try:
            parsed = RootCause.model_validate_json(
                body["output"][0]["content"][0]["text"]
            )
        except (ValidationError, KeyError, json.JSONDecodeError):
            continue                        # schema miss: retry, it may be transient
        return Result("ok", parsed.model_dump(), cost, attempt)

    return Result("invalid", None, cost, 3)

Three categories, three behaviors. Transport and rate-limit errors are free retries with exponential backoff. Schema violations are paid retries, capped at two, because temperature-zero output that fails validation twice usually fails forever on that input. Hard 4xx errors are terminal and get written with status = 'error' so the anti-join picks them up on the next run without blocking the batch.

Critically, the failure is recorded as a row, not raised. A single bad input should never kill a 40,000-row task. I write the batch in chunks of 500 and commit each chunk, so an infrastructure failure at row 31,000 loses at most 500 rows of work.

For the orchestration side, an LLM batch job is exactly the workload deferrable operators were built for: submit the batch, release the worker slot, poll for completion in the triggerer. A synchronous task holding a slot for six hours is a waste of a worker.

Quality: the part that has no exception to catch

Schema validation tells you the response was well-formed. It tells you nothing about whether billing was the right label. That failure mode produces a perfectly valid row, passes every dbt test you have, and corrupts a dashboard.

You need three layers, and they are cheap.

A pinned labeled sample. Three hundred rows, labeled by a human who knows the domain, stored in the repo as a CSV. Every prompt change, model change, and provider change runs against it in CI and reports accuracy per class. My rule: a prompt change that drops overall accuracy or drops any single class below 0.80 does not merge.

def evaluate(prompt: str, sample: list[tuple[str, str]]) -> dict:
    """Return per-class recall plus overall accuracy for a prompt."""
    hits, per_class = 0, {}
    for text, expected in sample:
        got = enrich_one(client, text, prompt)
        ok = got.status == "ok" and got.payload["root_cause"] == expected
        hits += ok
        c = per_class.setdefault(expected, [0, 0])
        c[0] += ok
        c[1] += 1
    return {
        "accuracy": hits / len(sample),
        "recall": {k: v[0] / v[1] for k, v in per_class.items()},
        "cost_usd": round(sum_cost(), 4),
    }

Distribution monitoring in production. The label distribution is a stable signal, and a shift in it is the earliest warning you get that something changed under you. Providers update models behind stable aliases; your prompt did not change but your outputs did.

-- Alert when any class share moves more than 5 points week over week.
with daily as (
    select date_trunc('week', created_at) as wk,
           root_cause,
           count(*) / sum(count(*)) over (partition by date_trunc('week', created_at))
               as share
    from analytics.enrich.ticket_root_cause
    where status = 'ok'
    group by 1, 2
)
select wk, root_cause, share,
       lag(share) over (partition by root_cause order by wk) as prev_share,
       share - lag(share) over (partition by root_cause order by wk) as delta
from daily
qualify abs(delta) > 0.05
order by wk desc;

A human sample loop. Twenty rows a week, reviewed by someone who knows the domain, recorded with the same schema as the labeled sample. This is how the labeled sample grows, and it is the only mechanism I have found that catches the slow drift where the model is subtly wrong on a category the eval set underrepresents.

Pin the model version explicitly. gpt-5.2 is an alias; gpt-5.2-2026-03 is a contract. Treating the model ID as part of the enrichment key, as in the table above, means a provider upgrade is a deliberate backfill you price and evaluate — the same discipline as a data contract version bump.

Where teams get this wrong

Idempotency keyed on the row ID. If the ticket body is edited, the ID is unchanged and you serve a stale label forever. Key on the hash of the exact input text you send.

No prompt version in the key. Change the prompt and you now have a table where half the rows were produced by one classifier and half by another, with no way to tell which. Every analysis over that table is quietly invalid.

Raising on the first bad row. One 90,000-character log dump should not cost you a 40,000-row batch. Record the failure, continue, and let the anti-join retry it tomorrow.

Using the synchronous API for batch work. You are paying double for latency the DAG does not need, and holding a worker slot while you do it.

Testing only that the JSON parses. Schema validation catches malformed output. It does not catch wrong output, which is the failure that actually reaches a dashboard. Pin a labeled sample.

Writing enrichment directly into the gold mart. Keep the raw enrichment table separate and join to it. When you need to reprocess with a new prompt, you want the old answers intact for comparison, not overwritten.

FAQ

Should the LLM call live in the warehouse or in Python?

In-warehouse functions like Snowflake’s AI_COMPLETE are excellent when the transformation fits a SQL statement and the corpus is moderate — no data egress, no separate service, and it composes with streams and tasks. I move to Python when I need batch-API pricing, custom retry logic, or a provider the warehouse does not host.

How do I make a non-deterministic step reproducible?

You do not make the model deterministic; you make the pipeline deterministic by caching. Temperature zero reduces variance but does not eliminate it. The content-hash table is what gives you reproducibility: the same input always yields the same stored answer because you never call the API twice for it.

Is it worth distilling to a small local model?

At high volume, often yes. Once you have 50,000 labeled outputs from a large model, fine-tuning a small open-weight classifier on them frequently matches accuracy at a fraction of the marginal cost. The break-even in my experience is somewhere north of a million rows a month; below that, the engineering time costs more than the API.

How do I handle rows the model refuses?

Store them with status = 'refused' and exclude them from the retry anti-join. Refusals are usually a signal about the input — PII, abuse content, an empty body — and retrying them just spends money to get the same answer. Report the refusal rate; a rising one means your input filter needs work.

What belongs in the dbt tests for an enrichment table?

Uniqueness on the composite key, an accepted-values test on the label column, a not-null test on status, and a freshness check. Then add a custom test that fails if the share of status != 'ok' exceeds a few percent. The general patterns are in data quality checks in Python.

How do I stop a runaway backfill?

A hard row cap and a hard dollar cap in the task itself, both read from config. My enrichment task refuses to process more than 50,000 rows or spend more than $75 in a single run, and pages instead. I have hit that cap twice, and both times it was a bug that would otherwise have cost four figures.

What this means for your pipelines

The uncomfortable realization is that almost nothing about LLM enrichment is new. Content-hash caching is what you already do for expensive geocoding APIs. Version-keyed outputs are what incremental models already need. Cost per row is what anyone who has run a Spark job on spot instances already thinks about. Sampling output for human review is what every classification system has always required.

What is new is the failure mode: a step that produces plausible, well-formed, expensive garbage without raising anything. Your existing tooling is built to detect absence and malformation, not confident wrongness. That gap is where the labeled sample, the distribution monitor, and the weekly human review earn their cost, and they are the three things teams skip because none of them are required to ship.

Build the enrichment table before you tune the prompt. Once the anti- join, the cost columns, and the eval harness exist, prompt iteration becomes a cheap, measurable, reversible activity instead of an unpriced experiment that quietly rewrites history. The prompt is the part everyone wants to work on. The table is the part that makes the prompt safe to change.