What Does One Successful AI Workflow Cost? Measuring LLM Spend in Django (Part 5)

Short version

• Price per token is the wrong number. What you pay for is a finished job: a cart the customer confirmed, a ticket that got resolved, a photo verdict dispatch didn't have to redo. Divide all LLM spend for a workflow type, failures, retries and abandoned runs included, by the number of successes.
• Success is a business event, not a finish_reason. The model saying "stop" proves nothing. Mark a workflow succeeded from the code path where the user confirms, with evidence attached.
• Prompt caching has three prices, not one. Uncached input, cache writes (1.25× base on Anthropic's 5-minute cache and OpenAI's 30-minute default for GPT-5.6+), and cache reads (usually 0.1×). A cold cache write costs more than not caching at all. Your ledger needs all three buckets.
• Budgets must fail closed, before the call. Check spent + worst_case(input, max_tokens) against the workflow budget under a row lock. Unknown model price? Refuse. Don't log a $0 call.
• The metric has a known blind spot. Cost per success rewards cheap models that only attempt easy tasks and quit early (Stet, Nov 6). Report it next to success rate and coverage, never on its own.
• 16 pytest tests cover cache-aware pricing, the budget guard, routing with escalation, success attribution and OTel attributes.

Say your LLM bill is $1,840 this month and your dashboard says the average request costs under a cent. Neither number tells you whether the AI feature is worth keeping.

The question a product owner actually asks is: what does it cost each time this thing works? If a cart assistant runs 1,000 times and 400 customers confirm the cart, a useful outcome costs 2.5× the per-run average. If a retry quietly reruns a failed photo check on a bigger model, that spend belongs to the same delivery, not a "misc tokens" bucket.

This is Part 5 of Shipping AI Agents in Django. Part 1 added human approval gates. Part 2 gave each agent run a scoped, short-lived grant. Part 3 put the same permission door behind MCP. Part 4 made retrieval respect row-level access. Those were about what the agent may do. This one is about what it costs, and how to stop it spending money you didn't approve.

Opinion, labelled as such: if you can't say what one successful outcome costs, you can't make a routing decision, a caching decision or a pricing decision. You're guessing.

The metric, and where it comes from

The definition that keeps showing up is simple. Arize and Fireworks used it in a July 2026 benchmark: total spend across every attempt (failed, retried, timed out) divided by the number of successful runs (Arize, repo). They ran 10 models on 40 Terminal-Bench tasks, 6 trials each (2,400 runs). gpt-oss-120b passed only 33% of the time yet cost $0.054 per success, against $0.636 for GPT-5.5 at 67%. gemini-3.5-flash cost $0.288 per attempt, passed 23% of the time, and landed at $1.233 per success, the worst in the table.

Two lessons for app developers:

1. Retries are part of the price. Three attempts per success means paying the input bill three times.
2. Cheap per token is not cheap per result. Measure on your workflow, with your definition of success. Grain64's write-up of the benchmark adds a sensible rule: write the pass rule before you run anything, and count the calls your validator rejected.

Benchmarks give you the shape. Production needs a ledger.

The data model: workflows own calls, outcomes own success

Two tables. A Workflow is one attempt at a business outcome: build a cart, verify a delivery photo, answer a support question. An LLMCall is one provider request inside it, with the token buckets split the way providers bill them.

class Workflow(models.Model):
    class Status(models.TextChoices):
        RUNNING = "running"
        SUCCEEDED = "succeeded"
        FAILED = "failed"

    kind = models.CharField(max_length=40, db_index=True)  # "cart_build", ...
    user = models.ForeignKey(settings.AUTH_USER_MODEL, null=True, on_delete=models.SET_NULL)
    status = models.CharField(max_length=12, choices=Status.choices, default=Status.RUNNING)
    budget_usd = models.DecimalField(max_digits=10, decimal_places=6)
    spent_usd = models.DecimalField(max_digits=12, decimal_places=6, default=Decimal("0"))
    failure_reason = models.CharField(max_length=40, blank=True)
    success_evidence = models.CharField(max_length=200, blank=True)  # e.g. "cart:LD-8120 confirmed"
    started_at = models.DateTimeField(auto_now_add=True)
    finished_at = models.DateTimeField(null=True)


class LLMCall(models.Model):
    workflow = models.ForeignKey(Workflow, on_delete=models.CASCADE, related_name="calls")
    step = models.CharField(max_length=40)
    attempt = models.PositiveSmallIntegerField(default=1)
    model = models.CharField(max_length=60)
    uncached_input_tokens = models.PositiveIntegerField(default=0)
    cache_read_tokens = models.PositiveIntegerField(default=0)
    cache_write_tokens = models.PositiveIntegerField(default=0)
    output_tokens = models.PositiveIntegerField(default=0)
    cost_usd = models.DecimalField(max_digits=12, decimal_places=6)
    ok = models.BooleanField(default=True)  # did OUR validator accept the output?
    created_at = models.DateTimeField(auto_now_add=True)

Three choices:

• budget_usd lives on the workflow, not the user. Per-user daily caps help, but one runaway agent loop can burn a day's cap in a single task. A per-workflow ceiling stops the loop.
• ok on each call is your validator's verdict, not the provider's. A response that parses but fails your schema is a failed attempt you paid for.
• success_evidence is required to mark success. It holds something like cart:LD-8120 confirmed. If you can't name the evidence, you don't have a success. You have a model that stopped talking.

Money is a six-place DecimalField. A cheap call costs fractions of a cent, and float rounding across millions of rows is a reconciliation headache you don't need.

Pricing: three input buckets, and the providers disagree on what "input" means

This is the part most hand-rolled cost trackers get wrong. As of today (verified 2026-11-09):

• Anthropic prices cache writes at 1.25× base input for the 5-minute TTL, 2× for the 1-hour TTL, and cache reads at 0.1× (0.05× on Opus 5.5 and Sonnet 5.5). Sonnet 5.5 is $2/MTok input, $2.50 for a 5-minute cache write, $0.10 for a cache hit, $10 output. Haiku 5.5, for prompts up to 100,000 tokens, is $0.10 / $0.125 / $0.01 / $0.50, and a different price tier applies above that (Anthropic prompt caching docs).
• OpenAI, for GPT-5.6 and later, charges cache writes at 1.25× uncached input for the default 30-minute lifetime, 2.5× for the 12-hour option, and reads at 0.1× on most of those models (OpenAI prompt caching guide).
• Google turns on implicit caching by default for Gemini 2.5 and newer, with minimum prompt sizes of 2,048 to 4,096 tokens depending on the model, and reports hits in the usage object (Gemini caching docs).

The trap is in the usage fields. Anthropic's input_tokens counts only the tokens after your last cache breakpoint. Total input is cache_read_input_tokens + cache_creation_input_tokens + input_tokens. OpenAI's Responses API input_tokens includes cached and cache-write tokens, so you subtract to get the ordinary portion. Add the two naively and one provider's cache reads get billed at full price in your ledger.

Normalise once, at the edge:

@dataclass(frozen=True)
class Usage:
    """One shape for every provider. ``total_input`` follows the OTel GenAI
    convention: it INCLUDES cache reads and cache writes."""

    uncached_input: int = 0
    cache_read: int = 0
    cache_write: int = 0
    output: int = 0

    @property
    def total_input(self) -> int:
        return self.uncached_input + self.cache_read + self.cache_write

    @classmethod
    def from_anthropic(cls, u: dict) -> "Usage":
        # Anthropic: input_tokens is only what came AFTER the last breakpoint.
        return cls(
            uncached_input=u.get("input_tokens", 0),
            cache_read=u.get("cache_read_input_tokens", 0) or 0,
            cache_write=u.get("cache_creation_input_tokens", 0) or 0,
            output=u.get("output_tokens", 0),
        )

    @classmethod
    def from_openai(cls, u: dict) -> "Usage":
        # OpenAI Responses: input_tokens INCLUDES cached and cache-write tokens.
        d = u.get("input_tokens_details") or {}
        cached = d.get("cached_tokens", 0) or 0
        written = d.get("cache_write_tokens", 0) or 0
        return cls(
            uncached_input=u["input_tokens"] - cached - written,
            cache_read=cached,
            cache_write=written,
            output=u.get("output_tokens", 0),
        )

total_input follows the OpenTelemetry GenAI convention: gen_ai.usage.input_tokens SHOULD include cached tokens, with cache_read and cache_write counts as subsets (semconv-genai). The attributes are still marked Development, so pin the version you emit. Matching them means traces and ledger agree on what "input tokens" means.

The price card refuses to guess:

def price_for(model: str, prompt_tokens: int = 0) -> PriceCard:
    try:
        card = PRICES[model]
    except KeyError:
        raise UnknownModelPrice(model) from None
    if card.max_prompt_tokens and prompt_tokens > card.max_prompt_tokens:
        # A different tier applies; we don't have it, so refuse.
        raise UnknownModelPrice(f"{model} over {card.max_prompt_tokens} tokens")
    return card


def cost_usd(model: str, usage: Usage) -> Decimal:
    p = price_for(model, usage.total_input)
    raw = (
        usage.uncached_input * p.input
        + usage.cache_read * p.cache_read
        + usage.cache_write * p.cache_write
        + usage.output * p.output
    ) / MTOK
    return raw.quantize(MICRO)

An unknown model raises instead of recording $0, because a $0 row makes a new model look free until the invoice arrives. The Haiku card carries max_prompt_tokens=100_000, so a longer prompt (a tier we haven't entered) is refused rather than under-charged.

What caching does to one workflow (worked example)

Take a cart-building step with a 6,000-token stable prefix (system prompt plus tool definitions), 200 tokens of new user input and 300 output tokens on Sonnet 5.5. These are illustrative inputs, not measurements. The prices come from the Anthropic table above:

The first cached call costs more than not caching. One write plus one read ($0.0224) already beats two uncached calls ($0.0308). After that, output tokens dominate the bill: $0.003 of the warm $0.004 is output. That's why a cache hit ratio alone tells you little, and why the demo reports it next to cost per success rather than instead of it.

Caching only pays when the prefix is byte-identical and still alive (Anthropic's default is 5 minutes, refreshed on each hit). Both Anthropic and OpenAI document the same trap: a breakpoint on content that changes every request, such as a timestamp, means a fresh write every call and no reads. If your ledger shows cache_write_tokens climbing and cache_read_tokens flat, that's the bug.

The budget guard: fail closed, before the network call

def metered_call(
    workflow: Workflow,
    *,
    step: str,
    model: str,
    input_tokens: int,
    max_output_tokens: int,
    call: Callable[[], ProviderResult],
    validate: Callable[[str], bool] = lambda _t: True,
    attempt: int = 1,
) -> tuple[ProviderResult, bool]:
    """Check budget, call the provider, record the real cost.

    Fails closed: unknown model price -> UnknownModelPrice before any call;
    worst case over budget -> BudgetExceeded before any call.
    """
    ceiling = worst_case_usd(model, input_tokens, max_output_tokens)  # may raise
    over = None
    with transaction.atomic():
        wf = Workflow.objects.select_for_update().get(pk=workflow.pk)
        if wf.status != Workflow.Status.RUNNING:
            raise WorkflowClosed(wf.status)
        if wf.spent_usd + ceiling > wf.budget_usd:
            wf.status = Workflow.Status.FAILED
            wf.failure_reason = "budget"
            wf.finished_at = timezone.now()
            wf.save(update_fields=["status", "failure_reason", "finished_at"])
            over = f"{wf.spent_usd} + {ceiling} > {wf.budget_usd}"
    if over:  # raise AFTER commit, or the failure mark rolls back with it
        raise BudgetExceeded(over)

    result = call()  # network call happens outside the row lock
    ok = bool(validate(result.text))
    cost = cost_usd(model, result.usage)
    with transaction.atomic():
        LLMCall.objects.create(
            workflow_id=workflow.pk,
            step=step,
            attempt=attempt,
            model=model,
            uncached_input_tokens=result.usage.uncached_input,
            cache_read_tokens=result.usage.cache_read,
            cache_write_tokens=result.usage.cache_write,
            output_tokens=result.usage.output,
            cost_usd=cost,
            ok=ok,
        )
        Workflow.objects.filter(pk=workflow.pk).update(spent_usd=F("spent_usd") + cost)
    return result, ok

What it does, in order:

1. Compute a worst-case ceiling before anything else. Every input token priced at the most expensive input rate (a cache write), plus the full max_output_tokens. If the model has no price card, this raises and no call happens.
2. Lock the workflow row with select_for_update() and check spent + ceiling <= budget. Two parallel agent steps can't both squeeze under the line.
3. Commit the failure, then raise. If the raise sits inside transaction.atomic(), the exception rolls back the status=failed write, so the workflow stays running with nothing to show for it. Mark it, commit it, then raise.
4. Call the provider outside the lock. Holding a row lock across a multi-second network call is how you get lock timeouts under load.
5. Record the actual cost from the provider's usage and add it with an F() expression, so two stale model instances can't overwrite each other's spend. One test does exactly that.

Trade-off: the worst case is pessimistic. A workflow with $0.02 left can be refused for a call that would really cost $0.004, because max_tokens=4000 could cost $0.04. That's the price of failing closed. If false refusals hurt, lower max_output_tokens per step rather than loosening the guard. It also caps how long a confused model can ramble.

Success attribution: from the confirm button, with evidence

def mark_succeeded(workflow: Workflow, evidence: str) -> None:
    """Call this from the business event (user confirmed the cart, ticket
    resolved), never from the model's finish_reason."""
    if not evidence:
        raise ValueError("success needs evidence")
    updated = Workflow.objects.filter(
        pk=workflow.pk, status=Workflow.Status.RUNNING
    ).update(status=Workflow.Status.SUCCEEDED, success_evidence=evidence, finished_at=timezone.now())
    if not updated:
        wf = Workflow.objects.get(pk=workflow.pk)
        if wf.status != Workflow.Status.SUCCEEDED:
            raise WorkflowClosed(wf.status)

Call mark_succeeded from the view that handles "Confirm cart" or the webhook that closes a ticket, never from the LLM wrapper. It's idempotent (a double-tap doesn't error), it won't flip a failed workflow to succeeded, and it refuses empty evidence. Workflows that time out or get abandoned should be closed by a periodic task with mark_failed(wf, "abandoned"). Otherwise they sit in running forever and stay out of the denominator, which flatters your number.

The report: all spend over successes

def report(kind: str, since=None) -> KindReport:
    wf = Workflow.objects.filter(kind=kind).exclude(status=Workflow.Status.RUNNING)
    if since is not None:
        wf = wf.filter(started_at__gte=since)
    agg = wf.aggregate(
        n=Count("id"),
        ok=Count("id", filter=Q(status=Workflow.Status.SUCCEEDED)),
        spend=Sum("spent_usd"),
        failed_spend=Sum("spent_usd", filter=~Q(status=Workflow.Status.SUCCEEDED)),
    )
    spend = agg["spend"] or Decimal("0")
    tok = LLMCall.objects.filter(workflow__in=wf).aggregate(
        read=Sum("cache_read_tokens"), write=Sum("cache_write_tokens"), un=Sum("uncached_input_tokens")
    )
    total_in = sum(v or 0 for v in tok.values())

The key line is failed_spend: failed and abandoned workflows are charged to the survivors. In the test, four cart workflows cost $0.004 each and two succeed, so cost per success is $0.008. Zero successes returns None, not zero. A division that silently returns 0 is how a broken feature ends up looking like the cheapest one on the dashboard.

Routing: cheapest first, escalate on visible failure

def choose_model(kind: str, *, input_tokens: int, attempt: int) -> str:
    ladder = settings.LLM_ROUTES[kind]
    if input_tokens > settings.LLM_ROUTE_LONG_CONTEXT_TOKENS:
        return ladder[-1]
    return ladder[min(attempt - 1, len(ladder) - 1)]


def routed_call(workflow: Workflow, *, step, input_tokens, max_output_tokens, call_for, validate):

The route table lives in settings: photo_check and support_reply start on Haiku 5.5 and escalate to Sonnet 5.5; cart_build goes straight to Sonnet. Long prompts skip the cheap rung entirely. routed_call meters every attempt into the same workflow, so the escalation test shows the Haiku attempt (rejected by the validator) and the Sonnet attempt (accepted) both landing on one delivery: $0.000175 + $0.003500 = $0.003675.

Here's the honest catch, and it's the most important paragraph in this post. Escalation only fires on failures your validator can see. Stet replayed a cheap-first router over recorded coding-agent attempts. Escalating only on visible failures (no patch, unfinished run) cut spend from $26.40 to $6.89. It also solved 8 of 24 tasks instead of 13 and shipped more failing patches, because a patch that fails hidden tests looks like a success to the router (Stet). For a photo verdict with three allowed values, a strict validator catches most nonsense. For free-text support replies, it won't. Route cheap only where your validator is strong.

Where cost per success lies to you

The same Stet post makes a case against using this metric alone, and it's worth taking seriously. On their 24-task board, ranking by cost per success put GPT-6 Luna low first at $0.26, even though it lost every head-to-head pairing and finished last on their quality score. Three reasons:

• Survivorship bias. Each model's spend is divided by its own successes. Luna low solved 5 tasks, all small fixes most models solved, and went 0 for 14 on multi-file features.
• Cheap quitters win. A run that gives up after two turns spends almost nothing, so failures look free.
• Binary success flattens quality. A patch that passes tests can still be one nobody wants to merge.

Translated to Django workflows:

• Report cost per success, success rate and coverage together. If a cheaper route drops success from 80% to 55%, a lower cost per success is a worse product.
• Slice by workflow shape. Stet's fix was to route cheap models only to task slices where they matched the expensive one. In app terms: route photo_check on clear daylight photos to Haiku, not "all photo checks."
• Keep failure_reason honest. Budget refusals are failures you caused. Track them apart from validator and abandoned, or you'll tune budgets down until success collapses.

A real-world shape: one app, several AI workflows

OnWay is one of our showcase apps, a local delivery demo with customer, driver and dispatcher roles. Its public plain-text spec lays out exactly the situation this post is about: Claude Sonnet handles chat and voice ordering (building carts with a build_cart tool the customer must confirm), the AI support agent and dispatcher fleet summaries. Claude Haiku does the vision check on proof-of-delivery photos, with a maximum of two checks per delivery. Each feature has its own daily cap, and every AI request is logged with user, feature, model and tokens.

That's per-feature metering and routing already. What this post adds is the denominator: confirmed carts, clear photo verdicts, resolved tickets. A Haiku photo check and a Sonnet cart build differ in price and in what success means; a blended "AI cost per user" hides both.

Emitting it to your traces

The ledger row maps directly onto OTel GenAI span attributes:

def genai_usage_attributes(call: LLMCall, provider: str = "anthropic") -> dict:
    total_input = call.uncached_input_tokens + call.cache_read_tokens + call.cache_write_tokens
    attrs = {
        "gen_ai.provider.name": provider,
        "gen_ai.operation.name": "chat",
        "gen_ai.request.model": call.model,
        "gen_ai.usage.input_tokens": total_input,  # includes cached tokens
        "gen_ai.usage.output_tokens": call.output_tokens,
    }
    if call.cache_read_tokens:
        attrs["gen_ai.usage.cache_read.input_tokens"] = call.cache_read_tokens
    if call.cache_write_tokens:
        attrs["gen_ai.usage.cache_write.input_tokens"] = call.cache_write_tokens

No OpenTelemetry dependency: attach the dict to whatever span you already have. Part 6 goes deeper.

Tests that pin the behaviour

16 tests, SQLite in memory, well under a second. The one I'd copy first:

def test_budget_guard_blocks_before_provider_is_called(make_wf):
    wf = make_wf(budget="0.010000")
    p = FakeProvider()
    # worst case: 6200 * 2.50 + 4000 * 10 = 55,500 / 1e6 = $0.0555 > $0.01
    with pytest.raises(BudgetExceeded):
        metered_call(wf, step="plan", model=SONNET, input_tokens=6200,
                     max_output_tokens=4000, call=p(SONNET))
    assert p.calls == []
    wf.refresh_from_db()
    assert (wf.status, wf.failure_reason) == ("failed", "budget")

p.calls == [] is the whole point of this test: the provider was never called. If a refactor ever moves the check after the call, CI fails.

What this doesn't do

• Provider errors after billing. If the SDK raises mid-stream, metered_call records nothing. Some providers still bill partial output. Wrap the call and record what usage you got back, or reconcile against the provider's usage export each night.
• Price drift. The price card is a dict with an as_of date. In production, keep it in a table with effective-from dates so historical rows keep their historical price, and alert when a model in traffic has no card.
• Batch and tier discounts. Anthropic's docs say cache multipliers stack with batch discounts, long-context pricing and data residency. The demo models none of that.
• Quality. A confirmed cart isn't proof it was the right cart. Pair this with an eval gate.

Checklist for your PR template

### LLM cost-per-outcome checklist
- [ ] Every LLM call belongs to a Workflow; usage split into 4 buckets
- [ ] Anthropic vs OpenAI input semantics normalised (OTel style)
- [ ] Unknown model or price tier -> refuse, never $0
- [ ] Worst-case budget check BEFORE the call, under a row lock, committed before raising
- [ ] Success marked from the business event, with evidence; abandoned runs closed
- [ ] Report: cost/success AND success rate AND spend on failures
- [ ] Cheap-first routing only where the validator is strict

Open question for readers

Here's the trade-off I haven't settled. Cost per success gives a product owner one number and charges failures honestly, but it rewards cheap configurations that only win easy cases. Paired reporting (success rate and cost per attempt on separate axes, Stet-style) is more truthful, but harder to put in a budget meeting.

Which do you put in front of the person who owns the bill, and how do you stop the cheap-quitter effect? And on budgets: do you guard on worst-case max_tokens like this demo, accepting some false refusals, or on a p95 of observed spend, accepting the occasional overrun?

The series so far

1. Human-in-the-loop AI agents in Django: approval gates
2. Least-privilege agent identity in Django
3. Your MCP server is a second API
4. Permission-aware RAG without post-filters
5. Cost per successful workflow (this post)

Next, Part 6: tracing LLM calls with OpenTelemetry. What to put on the span, what to redact (Part 4's canary rule applies), and how to go from "that answer was bad" to the exact prompt, retrieval hits and cost that produced it.

The demo is a plain Django 5.2 app: costs/ (pricing, ledger, router, metrics, otel) and one test file. More of what we build is at lycore.com.