Personalize this lesson
Adapt explanations and teaching visuals to your background and preferred voice.
PolicyOps has a fictional $750 monthly generation budget. Its support assistant can return a stored answer, generate with a cached prefix, or fetch current order data before generating. Those paths answer similar-looking questions but produce different bills. Which one should the team optimize first?
After selecting concise cited replies and an estimated Batch rate row, our synthetic plan costs $23.906 per day, including nightly evaluation. Repeating it for 30 days gives $717.18, leaving $32.82. We'll find out why passing that forecast doesn't enforce a spending cap.
The workload contains 8,500 interactive queries and 2,000 offline evaluation calls per day. We assign 3,200 queries a stored-answer hit, while 1,800 policy queries and 500 access exceptions generate with assumed prefix reads. Stored-answer hits have zero answer-generation charges; prefix reads still have output charges. Calling both simply "cache hits" hides that difference.
We'll price supplied usage, aggregate a forecast by feature, and check a local release policy. All traffic, token counts, cache outcomes, and quality labels are fixtures. The code makes no network calls, validates no real answers, and installs no telemetry, runtime cap, or release gate. Its records represent cohorts of identical calls, not actual per-request spans or provider invoices.
The accounting boundary
Before pricing a request, classify what happened at the answer-generation boundary. A stored answer skips that call; a prefix read still generates new output. Separate delivery outcomes from scheduling and input categories:
| Decision | What executes? | What belongs in this ledger? |
|---|---|---|
| Semantic answer hit from previous evaluation | Stored evaluated answer returned | No generation charge; record avoided generation separately |
| Generated answer with provider prompt-cache hit | Model generates a new answer | Ordinary input, cache reads/writes if reported, and output charges |
| Generated answer without prompt-cache hit | Model processes input and generates a new answer | Ordinary input, any cache writes, and output charges |
| Offline evaluation batch | Model runs asynchronously outside interactive path | Batch-eligible token spend with completion SLA |
A stored-answer hit, whether exact or semantic, skips answer generation. Semantic lookup may still run an embedding model. A provider prompt cache reuses compatible prefix computation and generates a new response. Neither outcome alone establishes answer correctness or freshness.
Zero generation spend doesn't mean a free user interaction. Vector search queries, embeddings, tool execution, and host infrastructure cost money. Maintain distinct ledger columns for tool invocation fees and storage costs. If a model call fails with a network error or timeout, record its status as unresolved and reconcile with provider logs rather than recording zero tokens.
After authorization and cache eligibility, ask whether answer generation ran. For generated calls, normalize all reported input categories before pricing. A request can read an old prefix and write a new suffix in the same call; "no cache read" doesn't imply "no cache-write fee."
The token economics matrix and inference physics
Many text-model rate cards charge more for output than input, but there is no universal 3x-to-6x rule. The documented GPT-5.4 row we'll use has a 6x ratio. Rates depend on the model and service tier; a price ratio alone doesn't tell you which category dominates a workload.[1]
Inference helps explain why input and output work differ. It doesn't establish how a provider sets its margins or prove that prices equal hardware costs:
- Prefill: Processes known input positions in parallel within each model layer. It often has high arithmetic intensity and is compute-bound. Long prefills may be chunked instead of processed all at once.
- Decode: Ordinary autoregressive generation depends on earlier generated tokens. Small-batch dense-model decode is often memory-bandwidth bound. Batching can amortize weight reads; sparse models don't activate every parameter per token,[2] and speculative decoding can verify several proposed tokens together.[3] Context length, kernels, architecture, hardware, and scheduling affect the bottleneck.[4]
Keep billing categories distinct from those execution phases:
- Ordinary input: Input outside cache-read and separately priced cache-write categories.
- Cache reads and writes: Different input categories with model-specific prices. A discounted read may follow a more expensive write.
- Visible output tokens: Autoregressively generated tokens streamed back to the client application.
- Reasoning detail: Part of the provider's output total. Reasoning can occur before or between visible output and tool calls.
For OpenAI reasoning models, billed output can include reasoning and non-visible formatting tokens. A max_output_tokens stop can occur before visible text appears, so an incomplete answer can still cost money. The output limit includes reasoning; it isn't a visible-word limit.[5]
Use a dated rate card, not remembered prices
Checked on September 21, 2026, the GPT-5.4 model page still documents $2.50 ordinary input, $0.25 cached input, and $15.00 output per million tokens. We'll retain that older model's row for a worked budget, then price a current GPT-5.6 Terra cache-write/read sequence separately. This isn't a model recommendation. Rates need model, service tier, currency, region, effective date, and cache policy; regional processing can add a surcharge.[1]
Above 272K input tokens, the full session switches to long-context pricing: twice the input rates and 1.5 times the output rate, meaning $5.00 standard input, $0.50 cached input, and $22.50 output. Crossing 272,000 input tokens by a single token reprices the entire request, not just the overflow volume.[1]
OpenAI documents 50% lower Batch costs with a 24-hour completion window. We'll explicitly estimate this GPT-5.4 row at half its standard rates: $1.25 input, $0.125 cache reads, and $7.50 output per million. Verify the chosen model/endpoint and billing terms before using a forecast. Decimal arithmetic avoids binary floating-point price error; this lab's finite-decimal products fit its precision. Round for display, and reconcile provider rounding and adjustments at their billing boundary.[6]
Use provider usage receipts, not character counts. For this older OpenAI row, ordinary input equals total input minus cache reads. For models with separately priced writes, subtract both reads and writes; our normalized Usage will support that too. Providers use different field meanings: Anthropic's input_tokens excludes its read/write counts, so copying it as OpenAI's total input would undercount.[7]
Reasoning tokens sit inside provider-reported output_tokens. For GPT-5.4, output_tokens already includes hidden reasoning tokens billed at the output rate. The field output_tokens_details.reasoning_tokens explains part of that total; adding it again would double-bill your ledger.
Before running the first cell, split a 1,800-token fixture reporting 400 cache-read tokens and no writes: 1,400 ordinary tokens and 400 reads. Usage shape validation can't prove that a real cache hit occurred. The cells share one Python session.
from collections import defaultdict
from dataclasses import asdict, dataclass, replace
from decimal import Decimal, ROUND_HALF_UP
from hashlib import sha256
import json
MILLION = Decimal("1000000")
CENT = Decimal("0.01")
def valid_decimal(value: object) -> bool:
return type(value) is Decimal and value.is_finite() and value >= 0
def nonempty_text(value: object) -> bool:
return type(value) is str and bool(value.strip())
def count(value: object) -> bool:
return type(value) is int and value >= 0
def dollars(amount: Decimal) -> Decimal:
if not valid_decimal(amount):
raise ValueError("money must be a finite nonnegative Decimal")
return amount.quantize(CENT, rounding=ROUND_HALF_UP)
@dataclass(frozen=True)
class TokenRates:
input_per_1m: Decimal
cached_input_per_1m: Decimal
output_per_1m: Decimal
cache_write_per_1m: Decimal | None = None
def __post_init__(self) -> None:
if not all(valid_decimal(value) for value in (
self.input_per_1m, self.cached_input_per_1m, self.output_per_1m,
)):
raise ValueError("rates must be finite nonnegative Decimals")
if self.cache_write_per_1m is not None and not valid_decimal(self.cache_write_per_1m):
raise ValueError("cache-write price is unknown or a finite nonnegative Decimal")
@dataclass(frozen=True)
class RateCard:
rate_card_id: str
standard: TokenRates
batch: TokenRates | None
maximum_input_tokens: int = 272_000
minimum_input_tokens: int = 0
model_id: str = "gpt-5.4"
processing_region: str = "global"
currency: str = "USD"
context_window_tokens: int = 1_050_000
maximum_output_tokens: int = 128_000
def __post_init__(self) -> None:
if not all(nonempty_text(value) for value in (
self.rate_card_id, self.model_id, self.processing_region, self.currency,
)) or self.currency != "USD":
raise ValueError("need named model, region, rate card, and USD currency")
if type(self.standard) is not TokenRates or (self.batch is not None and type(self.batch) is not TokenRates):
raise ValueError("need validated rates; unknown Batch prices remain None")
bounds = (self.minimum_input_tokens, self.maximum_input_tokens, self.context_window_tokens, self.maximum_output_tokens)
if not all(count(value) for value in bounds) or not (
self.minimum_input_tokens <= self.maximum_input_tokens <= self.context_window_tokens
and 0 < self.maximum_output_tokens <= self.context_window_tokens
):
raise ValueError("invalid rate-tier or model token limits")
def rate_digest(card: RateCard) -> str:
# A content fingerprint, not a signature or proof of truthful provider prices.
payload = json.dumps(asdict(card), sort_keys=True, separators=(",", ":"), default=str)
return sha256(payload.encode()).hexdigest()
@dataclass(frozen=True)
class Usage:
input_tokens: int
cached_input_tokens: int
output_tokens: int
reasoning_tokens: int = 0
cache_write_tokens: int = 0
def __post_init__(self) -> None:
counts = (self.input_tokens, self.cached_input_tokens, self.output_tokens, self.reasoning_tokens, self.cache_write_tokens)
if any(type(count) is not int for count in counts):
raise ValueError("token counts must be integers")
if min(counts) < 0:
raise ValueError("token counts cannot be negative")
if self.cached_input_tokens + self.cache_write_tokens > self.input_tokens:
raise ValueError("read and write input categories must fit the total")
if not 0 <= self.reasoning_tokens <= self.output_tokens:
raise ValueError("reasoning tokens must be a subset of output tokens")
rate_card = RateCard(
rate_card_id="gpt-5.4-global-short-estimated-batch-2026-09-21",
standard=TokenRates(
input_per_1m=Decimal("2.50"),
cached_input_per_1m=Decimal("0.25"),
output_per_1m=Decimal("15.00"),
),
batch=TokenRates(
input_per_1m=Decimal("1.25"),
cached_input_per_1m=Decimal("0.125"),
output_per_1m=Decimal("7.50"),
),
)
print(rate_card.rate_card_id)
print(f"standard_input_per_1m=${rate_card.standard.input_per_1m}")
print(f"standard_cached_input_per_1m=${rate_card.standard.cached_input_per_1m}")
print(f"standard_output_per_1m=${rate_card.standard.output_per_1m}")
print(f"batch_input_per_1m=${rate_card.batch.input_per_1m}")
print(f"batch_cached_input_per_1m=${rate_card.batch.cached_input_per_1m}")
print(f"batch_output_per_1m=${rate_card.batch.output_per_1m}")gpt-5.4-global-short-estimated-batch-2026-09-21
standard_input_per_1m=$2.50
standard_cached_input_per_1m=$0.25
standard_output_per_1m=$15.00
batch_input_per_1m=$1.25
batch_cached_input_per_1m=$0.125
batch_output_per_1m=$7.50Price one generated answer
Our fictional public-policy-answer route requires policy citations. We don't generate or inspect an answer here. For "How long are revoked access tokens retained?", supply a cold receipt with 1,800 input tokens and 180 output tokens. Which category costs more: the one with the higher rate or the one with ten times as many tokens?
Ordinary input costs 1,800 × $2.50 / 1,000,000 = $0.004500. Output costs 180 × $15.00 / 1,000,000 = $0.002700. Together, the call costs $0.007200. This older rate card has no separate cache-write price: any input used to create a cache remains ordinary input, and the separately priced write count is zero. For a card with explicit write fees, use four mutually exclusive categories:
@dataclass(frozen=True)
class PricedUsage:
fresh_input_usd: Decimal
cached_input_usd: Decimal
output_usd: Decimal
cache_write_usd: Decimal = Decimal("0")
@property
def total_usd(self) -> Decimal:
return self.fresh_input_usd + self.cached_input_usd + self.cache_write_usd + self.output_usd
def price_usage(usage: Usage, card: RateCard = rate_card, *, batch: bool = False) -> PricedUsage:
if type(usage) is not Usage or type(card) is not RateCard or type(batch) is not bool:
raise ValueError("need validated usage, rate card, and a Boolean Batch flag")
if not card.minimum_input_tokens <= usage.input_tokens <= card.maximum_input_tokens:
raise ValueError("input is outside this rate card's context tier")
rates = card.batch if batch else card.standard
if rates is None:
raise ValueError("no approved rates for this service tier")
if usage.input_tokens + usage.output_tokens > card.context_window_tokens or usage.output_tokens > card.maximum_output_tokens:
raise ValueError("usage exceeds the declared model token limits")
if usage.cache_write_tokens and rates.cache_write_per_1m is None:
raise ValueError("cache writes require an explicit approved price")
fresh_input_tokens = usage.input_tokens - usage.cached_input_tokens - usage.cache_write_tokens
return PricedUsage(
fresh_input_usd=Decimal(fresh_input_tokens) * rates.input_per_1m / MILLION,
cached_input_usd=Decimal(usage.cached_input_tokens) * rates.cached_input_per_1m / MILLION,
output_usd=Decimal(usage.output_tokens) * rates.output_per_1m / MILLION,
cache_write_usd=(
Decimal(usage.cache_write_tokens) * rates.cache_write_per_1m / MILLION
if usage.cache_write_tokens else Decimal("0")
),
)
cold_policy_answer = Usage(input_tokens=1_800, cached_input_tokens=0, output_tokens=180)
cold_cost = price_usage(cold_policy_answer)
assert cold_cost.total_usd == Decimal("0.0072")
assert rate_card.standard.output_per_1m > rate_card.standard.input_per_1m
print(f"fresh_input=${cold_cost.fresh_input_usd:.6f}")
print(f"output=${cold_cost.output_usd:.6f}")
print(f"total=${cold_cost.total_usd:.6f}")
print(f"output_share={(cold_cost.output_usd / cold_cost.total_usd):.1%}")fresh_input=$0.004500
output=$0.002700
total=$0.007200
output_share=37.5%The output token rate is 6x higher, but this 1,800-input-token fixture spends 62.5% of its dollars on input because input volume is ten times larger. Both volume and per-category rates belong in your ledger.
Prefix reuse can reduce prefill work and time to first token; it doesn't remove new output generation. Fewer billable output tokens reduce this ledger's output charge. Neither change guarantees a particular latency improvement under contention. Measure first-token, inter-token, and end-to-end latency separately.
In OpenAI's published usage shape, output_tokens=1186 includes reasoning_tokens=1024. The other 162 aren't necessarily all visible text: non-visible formatting can also count. Bill 1,186 once. Our next fixture supplies an incomplete call with 1,024 reasoning-only output tokens; a receipt's count alone doesn't establish the stop reason or text the user saw.[5]
reasoning_answer = Usage(
input_tokens=1_800,
cached_input_tokens=0,
output_tokens=1_186,
reasoning_tokens=1_024,
)
incomplete_reasoning = Usage(
input_tokens=1_800,
cached_input_tokens=0,
output_tokens=1_024,
reasoning_tokens=1_024,
)
reasoning_cost = price_usage(reasoning_answer)
incomplete_cost = price_usage(incomplete_reasoning)
double_counted_output = Decimal("2210") * rate_card.standard.output_per_1m / MILLION
assert reasoning_cost.output_usd == Decimal("1186") * rate_card.standard.output_per_1m / MILLION
assert reasoning_cost.output_usd != double_counted_output
assert incomplete_cost.output_usd == Decimal("1024") * rate_card.standard.output_per_1m / MILLION
assert incomplete_cost.total_usd > Decimal("0")
print(f"total_output_tokens={reasoning_answer.output_tokens}")
print(f"reasoning_detail={reasoning_answer.reasoning_tokens}")
print(f"priced_output=${reasoning_cost.output_usd:.6f}")
print(f"double_counted_output=${double_counted_output:.6f}")
print(f"incomplete_hidden_output_tokens={incomplete_reasoning.output_tokens}")
print(f"incomplete_total=${incomplete_cost.total_usd:.6f}")total_output_tokens=1186
reasoning_detail=1024
priced_output=$0.017790
double_counted_output=$0.033150
incomplete_hidden_output_tokens=1024
incomplete_total=$0.019860The incomplete call still incurs the $0.004500 fresh-input charge plus $0.015360 for hidden thinking. Hidden tokens aren't free, and reasoning_tokens isn't a secondary billable bucket. Retain the detail field for root-cause diagnosis when investigating why an agent blew through its turn budget without producing an answer.
Check the context tier before pricing
At 272,000 input tokens with zero cache reads and zero output, standard short-context cost is $0.68. At 272,001 tokens, the entire input flips to the long-context rate row, costing $1.360005. A single extra token doubles the cost of the full prompt. A 300,000-input, 1,000-output call costs $1.5225 under that long-context tier.
long_card = RateCard(
rate_card_id="gpt-5.4-global-long-estimated-batch-2026-09-21",
standard=TokenRates(Decimal("5"), Decimal("0.50"), Decimal("22.50")),
batch=TokenRates(Decimal("2.50"), Decimal("0.25"), Decimal("11.25")),
maximum_input_tokens=1_050_000,
minimum_input_tokens=272_001,
)
assert price_usage(Usage(272_000, 0, 0)).total_usd == Decimal("0.68")
assert price_usage(Usage(272_001, 0, 0), long_card).total_usd == Decimal("1.360005")
long_cost = price_usage(Usage(300_000, 0, 1_000), long_card).total_usd
assert long_cost == Decimal("1.5225")
try:
price_usage(Usage(272_001, 0, 0))
except ValueError as error:
print(f"wrong_tier_rejected={error}")
else:
raise AssertionError("short-context card accepted a long-context request")
for invalid in [(-1, 0, 0, 0), (10, 11, 0, 0), (10, 0, 2, 3), (1.5, 0, 0, 0)]:
try:
Usage(*invalid)
except ValueError:
pass
else:
raise AssertionError(f"invalid usage accepted: {invalid}")
print(f"long_context_total=${long_cost:.6f}")
print("invalid_usage_cases_rejected=4")wrong_tier_rejected=input is outside this rate card's context tier
long_context_total=$1.522500
invalid_usage_cases_rejected=4The card checks its declared pricing range and model token limits. It doesn't verify the provider's effective limits or select a row automatically. A real receipt needs the actual model, service tier, processing region, and session pricing policy. In multi-turn work, price each call and track session growth; the model-specific full-session rule can affect more than the overflow tokens.
Prefix reuse is still a generated answer
Checked October 10, 2026, OpenAI documents a 1,024-visible-token minimum and 1.25x cache-write prices for GPT-5.6 and later. Reads cost 0.1x ordinary input on most of these models, but 0.05x on GPT-6.1 Sol. For earlier models including GPT-5.4, the minimum varies with request settings; there is no universal 2,048-token minimum or additional write price. Anthropic documents 1.25x five-minute writes and 2x one-hour writes. Reads use 0.1x ordinary input except 0.025x for Fable 5.1 and Mythos 5.1, and 0.05x for Opus 5.5 and Sonnet 5.5. Pin the model's actual read price rather than applying one provider-wide discount.[8][7]
OpenAI matches the rendered prefix, which includes roles, tools, schemas, and relevant settings, not merely the characters in one user string. Preserve reusable instructions and context before the varying suffix, and use the appropriate cache boundaries. Moving data must preserve instruction priority, security boundaries, and the task's meaning. Prefix reuse doesn't make stale evidence current.
A real cache read needs provider telemetry. Our 1,280-token read is supplied to demonstrate arithmetic, not observed from GPT-5.4 or guaranteed for a particular request configuration. With 1,800 total input tokens and no separately priced writes, ordinary input is 520 and output remains 180.
prefix_cached_policy_answer = Usage(
input_tokens=1_800,
cached_input_tokens=1_280,
output_tokens=180,
)
prefix_cost = price_usage(prefix_cached_policy_answer)
savings = cold_cost.total_usd - prefix_cost.total_usd
assert prefix_cost.total_usd == Decimal("0.004320")
assert prefix_cost.output_usd == cold_cost.output_usd
assert savings == Decimal("0.002880")
print(f"cold_generation=${cold_cost.total_usd:.6f}")
print(f"prefix_cached_generation=${prefix_cost.total_usd:.6f}")
print(f"saved_per_generated_answer=${savings:.6f}")
print(f"still_generated={prefix_cost.output_usd > 0}")cold_generation=$0.007200
prefix_cached_generation=$0.004320
saved_per_generated_answer=$0.002880
still_generated=TrueWhy does a provider prompt-cache hit still have an output charge?
Answer
It reuses processing for an exactly matching input prefix. The model still generates the new response, so output tokens remain billable.
A cache write is not an ordinary input token plus a second fee
For a current comparison, GPT-5.6 Terra lists global Standard prices of $2.00 ordinary input, $0.20 cache reads, and $12.00 output per million; cache writes cost $2.50 (1.25 × $2.00). These rates were checked September 21, 2026.[9][10]
OpenAI reports cache_write_tokens beside cached_tokens inside input_tokens_details. Both are subsets of its input total. Price ordinary input as total minus reads minus writes. A write's 1.25x price already includes its base input charge. Charging the same tokens at the ordinary rate too would double-count that base.[8]
Predict the first write and subsequent read. We supply two 1,800-input, 180-output receipts with an explicit 1,280-token reusable boundary and no suffix writes. The first costs more than no caching; the pair costs less. No API call or cache configuration runs here. The modern card's Batch prices are deliberately unknown, and it doesn't enter the older-model monthly plan.
terra_card = RateCard(
rate_card_id="gpt-5.6-terra-global-standard-short-2026-09-21",
model_id="gpt-5.6-terra",
standard=TokenRates(Decimal("2.00"), Decimal("0.20"), Decimal("12.00"), Decimal("2.50")),
batch=None,
)
# These names mirror OpenAI's receipt. Anthropic needs a different adapter.
write_receipt = {
"input_tokens": 1_800,
"input_tokens_details": {"cached_tokens": 0, "cache_write_tokens": 1_280},
"output_tokens": 180,
}
write_usage = Usage(
input_tokens=write_receipt["input_tokens"],
cached_input_tokens=write_receipt["input_tokens_details"]["cached_tokens"],
cache_write_tokens=write_receipt["input_tokens_details"]["cache_write_tokens"],
output_tokens=write_receipt["output_tokens"],
)
read_usage = replace(write_usage, cached_input_tokens=1_280, cache_write_tokens=0)
ordinary_usage = replace(write_usage, cache_write_tokens=0)
write_cost = price_usage(write_usage, terra_card)
read_cost = price_usage(read_usage, terra_card)
ordinary_cost = price_usage(ordinary_usage, terra_card)
pair_savings = ordinary_cost.total_usd * 2 - write_cost.total_usd - read_cost.total_usd
assert write_cost.fresh_input_usd == Decimal("0.001040")
assert write_cost.cache_write_usd == Decimal("0.003200")
assert write_cost.total_usd == Decimal("0.006400")
assert read_cost.total_usd == Decimal("0.003456")
assert pair_savings == Decimal("0.001664")
print(f"no_cache_call=${ordinary_cost.total_usd:.6f}")
print(f"first_write_call=${write_cost.total_usd:.6f}")
print(f"subsequent_read_call=${read_cost.total_usd:.6f}")
print(f"two_call_savings=${pair_savings:.6f}")no_cache_call=$0.005760
first_write_call=$0.006400
subsequent_read_call=$0.003456
two_call_savings=$0.001664If the prefix is never reused, the extra write charge buys no read savings. If a call both reads and writes, include both. Don't reuse this calculation for a different provider's receipt fields or infer cache availability from a matching string.
Carry forward the semantic-cache decision
The preceding lesson simulated a narrow reuse policy; it didn't approve a production cache. Here we assume valid stored-answer outcomes within that fictional scope. The ledger prices those outcomes; it doesn't check current permissions, review the answer, or rebuild the vector index.
For "How long are revoked access tokens retained?", a stored-answer hit records zero answer-generation tokens. Under the older worked card, a prefix hit records a generated answer with cheaper input. Predict the rows: $0 generation for the stored answer with a $0.004320 estimated counterfactual, versus $0.004320 estimated generation for the prefix hit. Neither is a real billing observation.
@dataclass(frozen=True)
class RequestOutcome:
feature: str
decision: str
usage: Usage | None
counterfactual_usage: Usage | None = None
def __post_init__(self) -> None:
if not nonempty_text(self.feature) or not nonempty_text(self.decision) or self.decision not in {
"SEMANTIC_ANSWER_HIT", "GENERATE_PREFIX_HIT", "GENERATE_LIVE_DATA", "GENERATE_FRESH",
}:
raise ValueError("need a known decision and feature")
stored_answer = self.decision == "SEMANTIC_ANSWER_HIT"
if stored_answer != (self.usage is None):
raise ValueError("only a stored-answer hit may omit generated usage")
if self.usage is not None and type(self.usage) is not Usage:
raise ValueError("generated usage must be validated")
if self.counterfactual_usage is not None and (not stored_answer or type(self.counterfactual_usage) is not Usage):
raise ValueError("this counterfactual belongs only to a stored-answer hit")
if self.decision == "GENERATE_PREFIX_HIT" and self.usage.cached_input_tokens == 0:
raise ValueError("a prefix-hit label needs a reported read")
if self.decision == "GENERATE_FRESH" and self.usage.cached_input_tokens:
raise ValueError("fresh-generation label disagrees with reported reads")
def generated_cost(outcome: RequestOutcome) -> Decimal:
if type(outcome) is not RequestOutcome:
raise ValueError("need a validated request outcome")
if outcome.usage is None:
return Decimal("0")
return price_usage(outcome.usage).total_usd
semantic_hit = RequestOutcome(
feature="public-policy-answer",
decision="SEMANTIC_ANSWER_HIT",
usage=None,
counterfactual_usage=prefix_cached_policy_answer,
)
prefix_hit = RequestOutcome(
feature="public-policy-answer",
decision="GENERATE_PREFIX_HIT",
usage=prefix_cached_policy_answer,
)
avoided_generation = price_usage(semantic_hit.counterfactual_usage).total_usd
assert generated_cost(semantic_hit) == Decimal("0")
assert generated_cost(prefix_hit) == Decimal("0.004320")
try:
RequestOutcome("public-policy-answer", "GENERATE_PREFIX_HIT", None)
except ValueError as error:
print(f"missing_usage_rejected={error}")
else:
raise AssertionError("missing generated usage was priced as zero")
print(f"semantic_hit_generation=${generated_cost(semantic_hit):.6f}")
print(f"semantic_hit_avoided=${avoided_generation:.6f}")
print(f"prefix_hit_generation=${generated_cost(prefix_hit):.6f}")missing_usage_rejected=only a stored-answer hit may omit generated usage
semantic_hit_generation=$0.000000
semantic_hit_avoided=$0.004320
prefix_hit_generation=$0.004320
Replay a day of traffic
Evaluating cost per request hides the distribution of real traffic. Reconstruct one synthetic PolicyOps day before testing candidate prompt optimizations:
- 3,200 stable policy questions receive supplied stored-answer-hit labels.
- 1,800 policy questions generate new answers with assumed prefix reads.
- 3,000 live-order questions generate from fresh database tool evidence in this replay.
- 500 access exceptions generate a cited response with the stable prefix.
Before running the replay, predict which slice dominates. Live-order answers use fresh tool evidence and no prefix reads in this fixture. Their instructions could still have a reusable prefix in a real system; dynamic evidence doesn't make the whole prompt uncacheable. Avoided generation uses the supplied prefix-hit alternative, not the cold alternative, and must stay separate from estimated actual generation.
@dataclass(frozen=True)
class TrafficSlice:
outcome: RequestOutcome
requests: int
def __post_init__(self) -> None:
if type(self.outcome) is not RequestOutcome or not count(self.requests):
raise ValueError("need a validated outcome and nonnegative integer volume")
daily_replay = [
TrafficSlice(semantic_hit, requests=3_200),
TrafficSlice(prefix_hit, requests=1_800),
TrafficSlice(
RequestOutcome(
feature="live-order-answer",
decision="GENERATE_LIVE_DATA",
usage=Usage(input_tokens=900, cached_input_tokens=0, output_tokens=95),
),
requests=3_000,
),
TrafficSlice(
RequestOutcome(
feature="access-exception-answer",
decision="GENERATE_PREFIX_HIT",
usage=Usage(input_tokens=2_200, cached_input_tokens=1_280, output_tokens=220),
),
requests=500,
),
]
actual_daily_spend = sum(
generated_cost(slice_.outcome) * slice_.requests
for slice_ in daily_replay
)
avoided_daily_generation = sum(
price_usage(slice_.outcome.counterfactual_usage).total_usd * slice_.requests
for slice_ in daily_replay
if slice_.outcome.counterfactual_usage is not None
)
assert actual_daily_spend == Decimal("21.761000")
assert avoided_daily_generation == Decimal("13.824000")
print(f"generated_daily_spend=${dollars(actual_daily_spend)}")
print(f"semantic_hit_avoided_daily=${dollars(avoided_daily_generation)}")
print(f"request_count={sum(slice_.requests for slice_ in daily_replay)}")generated_daily_spend=$21.76
semantic_hit_avoided_daily=$13.82
request_count=8500The daily replay estimates $21.76 in generated token spend and $13.82 in avoided generation. Avoided generation is an audit metric showing cache effectiveness; it isn't an invoice credit. If a cached answer later fails a compliance audit, that failure matters far more than the theoretical dollars saved.
Reduce output only under an answer contract
Output brevity is one possible lever. In our cold example, input costs more despite output's higher rate. Rank the avoidable dollars before choosing a change, and preserve information the reader needs.
PolicyOps's cited support answer must satisfy three contract requirements: state the eligibility decision, cite policy evidence, and name the requester's next step. Compare three candidate prompt formats for the access exception feature:
| Candidate | Intended change | Supplied format result |
|---|---|---|
| Verbose | Longer prose | Meets the three supplied labels |
| Concise cited | One decision sentence, one citation, one action link | Shortest format meeting those labels |
| Truncated | Token limit cuts off citation and action | Fails two required labels |
The code checks supplied Boolean labels, not candidate text. All three token counts are fixtures; presence of a citation doesn't prove it supports the decision. A real comparison needs factual correctness, citation validity, action clarity, and usefulness across representative cases. Which candidate survives the stated format requirements?
@dataclass(frozen=True)
class AnswerCandidate:
label: str
output_tokens: int
states_decision: bool
cites_policy: bool
gives_next_step: bool
def __post_init__(self) -> None:
if not nonempty_text(self.label) or not count(self.output_tokens) or self.output_tokens == 0:
raise ValueError("need a named answer and positive integer output count")
if any(type(value) is not bool for value in (self.states_decision, self.cites_policy, self.gives_next_step)):
raise ValueError("contract labels must be known Booleans")
def satisfies_answer_contract(candidate: AnswerCandidate) -> bool:
if type(candidate) is not AnswerCandidate:
return False
return (
candidate.states_decision
and candidate.cites_policy
and candidate.gives_next_step
)
candidates = [
AnswerCandidate("verbose", 220, True, True, True),
AnswerCandidate("concise_cited", 130, True, True, True),
AnswerCandidate("truncated", 55, True, False, False),
]
format_candidates = [candidate for candidate in candidates if satisfies_answer_contract(candidate)]
selected_answer = min(format_candidates, key=lambda candidate: candidate.output_tokens)
verbose_usage = Usage(input_tokens=2_200, cached_input_tokens=1_280, output_tokens=220)
concise_usage = Usage(input_tokens=2_200, cached_input_tokens=1_280, output_tokens=selected_answer.output_tokens)
saved_per_exception = price_usage(verbose_usage).total_usd - price_usage(concise_usage).total_usd
assert selected_answer.label == "concise_cited"
assert not satisfies_answer_contract(candidates[-1])
assert saved_per_exception == Decimal("0.001350")
print(f"selected={selected_answer.label}")
print(f"rejected={candidates[-1].label}")
print(f"saved_per_exception=${saved_per_exception:.6f}")
print(f"saved_per_replay_day=${dollars(saved_per_exception * 500)}")selected=concise_cited
rejected=truncated
saved_per_exception=$0.001350
saved_per_replay_day=$0.68The cheapest candidate fails two supplied requirements. The selected candidate saves $0.001350 per exception, or $0.675 before rounding across 500 calls. It preserves these labels; this isn't evidence of real answer quality. Measure p95 latency too: a shorter visible answer can still consume reasoning tokens or wait in a queue.
Batch only work that can wait
Batch processing is an architectural scheduling decision before it's a pricing discount. OpenAI documents an asynchronous Batch API with a 50% discount and a 24-hour completion window. Requests that don't complete within the window can expire, while completed calls remain billable. A live customer waiting on support can't accept that latency SLA. Offline evaluation fits cleanly, provided its operational deadline leaves ample time to inspect failures and resubmit incomplete work.[6]
The check uses supplied model/endpoint compatibility and reserves four hours for recovery beyond the provider's 24-hour window. That buffer is our policy, not an API guarantee. Live support stays interactive; nightly evaluation has a 48-hour deadline. Its older-model estimate is $2.82 with assumed reads or $5.70 without them. Batch doesn't guarantee cache hits or successful completion.
@dataclass(frozen=True)
class Workload:
name: str
interactive: bool
deadline_hours: int
endpoint_supported: bool
daily_requests: int
usage: Usage
model_supported: bool
def __post_init__(self) -> None:
if not nonempty_text(self.name) or type(self.usage) is not Usage:
raise ValueError("need a named workload and validated usage")
if not count(self.deadline_hours) or not count(self.daily_requests):
raise ValueError("deadline and volume must be nonnegative integers")
if any(type(value) is not bool for value in (self.interactive, self.endpoint_supported, self.model_supported)):
raise ValueError("scheduling and compatibility flags must be known Booleans")
def batch_eligible(workload: Workload, recovery_hours: int = 4) -> bool:
if type(workload) is not Workload or not count(recovery_hours):
return False
return (
not workload.interactive
and workload.deadline_hours >= 24 + recovery_hours
and workload.endpoint_supported
and workload.model_supported
)
live_support = Workload(
name="live-support-answer",
interactive=True,
deadline_hours=0,
endpoint_supported=True,
daily_requests=3_000,
usage=Usage(input_tokens=900, cached_input_tokens=0, output_tokens=95),
model_supported=True,
)
nightly_eval = Workload(
name="nightly-release-eval",
interactive=False,
deadline_hours=48,
endpoint_supported=True,
daily_requests=2_000,
usage=Usage(input_tokens=1_800, cached_input_tokens=1_280, output_tokens=80),
model_supported=True,
)
nightly_sync_cost = price_usage(nightly_eval.usage).total_usd * nightly_eval.daily_requests
nightly_batch_cost = price_usage(nightly_eval.usage, batch=True).total_usd * nightly_eval.daily_requests
assert not batch_eligible(live_support)
assert batch_eligible(nightly_eval)
assert nightly_batch_cost == Decimal("2.820000")
assert nightly_batch_cost < nightly_sync_cost
nightly_cold_cost = price_usage(
replace(nightly_eval.usage, cached_input_tokens=0), batch=True
).total_usd * nightly_eval.daily_requests
assert nightly_cold_cost == Decimal("5.70")
print(f"live_support_batchable={batch_eligible(live_support)}")
print(f"nightly_eval_batchable={batch_eligible(nightly_eval)}")
print(f"nightly_eval_sync=${dollars(nightly_sync_cost)}")
print(f"nightly_eval_batch=${dollars(nightly_batch_cost)}")
print(f"nightly_eval_batch_without_cache=${dollars(nightly_cold_cost)}")live_support_batchable=False
nightly_eval_batchable=True
nightly_eval_sync=$5.64
nightly_eval_batch=$2.82
nightly_eval_batch_without_cache=$5.70Your nightly evaluation workload is cheaper through Batch. Should the interactive support endpoint use the same path?
Answer
No. The Batch completion window changes latency semantics. Only work that can wait belongs on that path.
Attribute cost to decisions, not models alone
A billing statement records charges; retained receipts and traces help explain their causes. Connect each real attempt to its task, feature, release, effective model/tier/region, usage, status, and pricing evidence. Missing receipts stay unresolved rather than turning into zero spend. An evidence ID alone doesn't prove that review passed.
Our LedgerTrace is an aggregate fixture, not an OpenTelemetry span. Its requests field multiplies one supplied unit receipt by a cohort count. Real calls can have different lengths, reasoning, failed attempts, and cache outcomes, so collect actual receipts before aggregating. Evaluate cache safety separately; cost metadata can't detect an unsupported answer.
Pin the price table's contents as well as its name. Keeping the same ID while changing rates would otherwise silently reprice history. A digest catches that content change against a trusted recorded fingerprint. It doesn't authenticate the table or retain it: archive approved rate cards and source evidence, and restrict who can publish or reprice them.
Rank estimated dollars by feature before refactoring prompts. In this synthetic day, live-order answers account for $11.03 out of $23.91, or 46.1% of total spend. If a feature crosses a team triage threshold (like 40%), audit its input prefill, tool payload size, and schema verbosity first.
@dataclass(frozen=True)
class LedgerTrace:
release_id: str
feature: str
decision: str
requests: int
usage: Usage | None
rate_card_id: str
rate_card_digest: str
model_id: str
contract_passed: bool
contract_evidence_id: str
batch: bool = False
def __post_init__(self) -> None:
if not all(nonempty_text(value) for value in (
self.release_id, self.feature, self.decision, self.rate_card_id, self.rate_card_digest,
self.model_id, self.contract_evidence_id,
)) or not count(self.requests):
raise ValueError("trace needs owned identifiers and nonnegative integer volume")
if type(self.contract_passed) is not bool or type(self.batch) is not bool:
raise ValueError("trace flags must be known Booleans")
if self.decision == "BATCH_OFFLINE_EVAL":
if not self.batch or type(self.usage) is not Usage:
raise ValueError("Batch trace needs validated generated usage")
else:
if self.batch:
raise ValueError("interactive cohort cannot claim Batch rates")
RequestOutcome(self.feature, self.decision, self.usage)
def trace_unit_spend(trace: LedgerTrace, card: RateCard = rate_card) -> Decimal:
if type(trace) is not LedgerTrace or type(card) is not RateCard:
raise ValueError("need a validated trace and card")
if trace.rate_card_id != card.rate_card_id:
raise ValueError("trace rate card does not match pricing table")
if trace.rate_card_digest != rate_digest(card) or trace.model_id != card.model_id:
raise ValueError("trace pricing content or model does not match")
if trace.usage is None:
if trace.decision != "SEMANTIC_ANSWER_HIT":
raise ValueError("generated trace has unresolved usage")
return Decimal("0")
return price_usage(trace.usage, card, batch=trace.batch).total_usd
def trace_spend(trace: LedgerTrace, card: RateCard = rate_card) -> Decimal:
return trace_unit_spend(trace, card) * trace.requests
release_id = "support-release-2026-05-cost-v1"
optimized_replay = [
TrafficSlice(
RequestOutcome(
feature=item.outcome.feature,
decision=item.outcome.decision,
usage=concise_usage if item.outcome.feature == "access-exception-answer" else item.outcome.usage,
counterfactual_usage=item.outcome.counterfactual_usage,
),
requests=item.requests,
)
for item in daily_replay
]
traces = [
LedgerTrace(
release_id=release_id,
feature=item.outcome.feature,
decision=item.outcome.decision,
requests=item.requests,
usage=item.outcome.usage,
rate_card_id=rate_card.rate_card_id,
rate_card_digest=rate_digest(rate_card),
model_id=rate_card.model_id,
contract_passed=True,
contract_evidence_id=(
"fixture-public-policy-cache@v1"
if item.outcome.decision == "SEMANTIC_ANSWER_HIT"
else "fixture-cited-support-answer-v3"
),
)
for item in optimized_replay
]
traces.append(
LedgerTrace(
release_id=release_id,
feature=nightly_eval.name,
decision="BATCH_OFFLINE_EVAL",
requests=nightly_eval.daily_requests,
usage=nightly_eval.usage,
rate_card_id=rate_card.rate_card_id,
rate_card_digest=rate_digest(rate_card),
model_id=rate_card.model_id,
contract_passed=True,
contract_evidence_id="fixture-nightly-eval@batch-eligible",
batch=True,
)
)
spend_by_feature: defaultdict[str, Decimal] = defaultdict(lambda: Decimal("0"))
for trace in traces:
spend_by_feature[trace.feature] += trace_spend(trace)
total_with_eval = sum(spend_by_feature.values())
trace_contracts_complete = all(
trace.contract_passed
and trace.contract_evidence_id
and trace.rate_card_id == rate_card.rate_card_id
and trace.rate_card_digest == rate_digest(rate_card)
and trace.release_id == release_id
for trace in traces
)
stale_trace = replace(traces[0], rate_card_id="openai-gpt-5.4-short-context-2026-04-30")
assert spend_by_feature["public-policy-answer"] == Decimal("7.776000")
assert actual_daily_spend - sum(trace_spend(trace) for trace in traces[:-1]) == Decimal("0.675000")
assert trace_contracts_complete
for feature in sorted(spend_by_feature):
print(f"{feature}=${dollars(spend_by_feature[feature])}")
print(f"daily_total_with_eval=${dollars(total_with_eval)}")
print(f"trace_contracts_complete={trace_contracts_complete}")
try:
trace_spend(stale_trace)
except ValueError as error:
print(f"stale_rate_card_rejected={error}")
else:
raise AssertionError("stale card accepted")
changed_rates_same_id = replace(rate_card, standard=replace(rate_card.standard, output_per_1m=Decimal("20")))
try:
trace_spend(traces[1], changed_rates_same_id)
except ValueError as error:
print(f"same_id_changed_prices_rejected={error}")
else:
raise AssertionError("changed prices passed the historical fingerprint")access-exception-answer=$2.29
live-order-answer=$11.03
nightly-release-eval=$2.82
public-policy-answer=$7.78
daily_total_with_eval=$23.91
trace_contracts_complete=True
stale_rate_card_rejected=trace rate card does not match pricing table
same_id_changed_prices_rejected=trace pricing content or model does not match
Forecast and gate the release
A production release gate answers two distinct questions: can we afford this workload, and can we trust the model's outputs? A budget gate alone rewards truncated, hallucinated answers. A quality gate alone risks deploying a runaway financial liability. Enforce both as independent prerequisites.
The next cell checks a supplied quality report, not a deployment. It derives cited-answer pass rate from reviewed integer counts and separately counts reviewed cache hits. It requires complete evidence for the current release/schema, report freshness, and our chosen sample minima. IDs and labels assume trusted ingestion; they aren't authenticated by this code.
These floors are lab policy, not a statistical proof of population quality. A real evaluation needs verified case membership, representative slices, uncertainty, and accurate review. Repeating $23.906 for 30 days gives $717.18 against $750.00. That forecast assumes fixed volumes and category distributions; it doesn't constrain live spending.
@dataclass(frozen=True)
class QualityReport:
report_id: str
release_id: str
answer_schema: str
cited_answers_passed: int
unsafe_cache_hits: int
evaluated_answers: int
cache_hits_reviewed: int
completed_at: int
def __post_init__(self) -> None:
if not all(nonempty_text(value) for value in (self.report_id, self.release_id, self.answer_schema)):
raise ValueError("quality report needs owned evidence and release/schema identity")
if not all(count(value) for value in (
self.cited_answers_passed, self.unsafe_cache_hits, self.evaluated_answers,
self.cache_hits_reviewed, self.completed_at,
)) or not (
self.evaluated_answers > 0
and self.cited_answers_passed <= self.evaluated_answers
and self.unsafe_cache_hits <= self.cache_hits_reviewed
):
raise ValueError("invalid evaluation counts")
@property
def cited_answer_pass_rate(self) -> Decimal:
return Decimal(self.cited_answers_passed) / Decimal(self.evaluated_answers)
@dataclass(frozen=True)
class PromotionPolicy:
monthly_budget_usd: Decimal
minimum_cited_answer_pass_rate: Decimal
maximum_unsafe_cache_hits: int
minimum_evaluated_answers: int
minimum_cache_hits_reviewed: int
required_release_id: str
required_answer_schema: str
maximum_report_age_seconds: int = 3_600
def __post_init__(self) -> None:
if not valid_decimal(self.monthly_budget_usd) or not valid_decimal(self.minimum_cited_answer_pass_rate) or self.minimum_cited_answer_pass_rate > 1:
raise ValueError("budget and quality floor must be finite nonnegative Decimals")
if not all(count(value) for value in (
self.maximum_unsafe_cache_hits, self.minimum_evaluated_answers,
self.minimum_cache_hits_reviewed, self.maximum_report_age_seconds,
)) or min(self.minimum_evaluated_answers, self.minimum_cache_hits_reviewed, self.maximum_report_age_seconds) == 0:
raise ValueError("need positive sample minima and evidence lifetime")
if not all(nonempty_text(value) for value in (self.required_release_id, self.required_answer_schema)):
raise ValueError("promotion policy needs release/schema identity")
EVIDENCE_NOW = 1_800_000_000 # Fixed lab clock, not an observed completion time.
quality = QualityReport(
report_id="fixture-quality-report-v1",
release_id=release_id,
answer_schema="cited-support-answer-v3",
cited_answers_passed=1_994,
unsafe_cache_hits=0,
evaluated_answers=2_000,
cache_hits_reviewed=1_200,
completed_at=EVIDENCE_NOW,
)
promotion_policy = PromotionPolicy(
monthly_budget_usd=Decimal("750.00"),
minimum_cited_answer_pass_rate=Decimal("0.995"),
maximum_unsafe_cache_hits=0,
minimum_evaluated_answers=2_000,
minimum_cache_hits_reviewed=1_000,
required_release_id=release_id,
required_answer_schema="cited-support-answer-v3",
)
monthly_forecast = total_with_eval * Decimal("30")
budget_headroom = promotion_policy.monthly_budget_usd - monthly_forecast
uniform_volume_growth_headroom = budget_headroom / monthly_forecast
sample_size_passed = (
quality.evaluated_answers >= promotion_policy.minimum_evaluated_answers
)
def passes_quality(report: QualityReport, policy: PromotionPolicy, evidence_complete: bool, now: int = EVIDENCE_NOW) -> bool:
if type(report) is not QualityReport or type(policy) is not PromotionPolicy or evidence_complete is not True or not count(now):
return False
return (
report.release_id == policy.required_release_id
and report.answer_schema == policy.required_answer_schema
and report.completed_at <= now <= report.completed_at + policy.maximum_report_age_seconds
and report.evaluated_answers >= policy.minimum_evaluated_answers
and report.cache_hits_reviewed >= policy.minimum_cache_hits_reviewed
and report.cited_answer_pass_rate >= policy.minimum_cited_answer_pass_rate
and report.unsafe_cache_hits <= policy.maximum_unsafe_cache_hits
)
quality_passed = passes_quality(quality, promotion_policy, trace_contracts_complete)
budget_passed = monthly_forecast <= promotion_policy.monthly_budget_usd
assert dollars(monthly_forecast) == Decimal("717.18")
assert dollars(budget_headroom) == Decimal("32.82")
assert quality_passed and budget_passed
assert not passes_quality(replace(quality, unsafe_cache_hits=1), promotion_policy, True)
assert not passes_quality(replace(quality, evaluated_answers=1_999), promotion_policy, True)
assert not passes_quality(quality, promotion_policy, False)
assert passes_quality(replace(quality, cited_answers_passed=1_990), promotion_policy, True)
assert not passes_quality(replace(quality, cache_hits_reviewed=999), promotion_policy, True)
foreign_release_passed = passes_quality(replace(quality, release_id="another-release"), promotion_policy, True)
stale_quality_passed = passes_quality(replace(quality, completed_at=EVIDENCE_NOW - 3_601), promotion_policy, True)
assert not foreign_release_passed and not stale_quality_passed
print(f"monthly_forecast=${dollars(monthly_forecast)}")
print(f"budget_headroom=${dollars(budget_headroom)}")
print(f"uniform_volume_growth_headroom={uniform_volume_growth_headroom:.1%}")
print(f"sample_size_passed={sample_size_passed}")
print(f"cited_answer_pass_rate={quality.cited_answer_pass_rate:.1%}")
print(f"cache_sample_size_passed={quality.cache_hits_reviewed >= promotion_policy.minimum_cache_hits_reviewed}")
print(f"foreign_release_passed={foreign_release_passed}")
print(f"stale_quality_passed={stale_quality_passed}")
print(f"quality_passed={quality_passed}")
print(f"budget_passed={budget_passed}")monthly_forecast=$717.18
budget_headroom=$32.82
uniform_volume_growth_headroom=4.6%
sample_size_passed=True
cited_answer_pass_rate=99.7%
cache_sample_size_passed=True
foreign_release_passed=False
stale_quality_passed=False
quality_passed=True
budget_passed=TrueThe $32.82 is 4.4% of the $750 budget, but permits 4.6% growth relative to the $717.18 forecast. Those denominators differ. Uniform growth assumes every cohort, including evaluation, scales together with unchanged token and cache distributions. A passing forecast with this cushion can still exceed the budget after retries or traffic changes.
Two routes have the same cost per attempt, but one often needs a retry before producing a valid answer. Which cost should the release decision compare?
Answer
Compare cost per successful answer or task, including retries and failed attempts. Cost per request would hide the extra spend caused by lower reliability.
Unit economics and the retry tax
Measuring "cost per request" or "cost per million tokens" is one of the most common pitfalls in LLMOps. Comparing models solely by token rate ignores model accuracy and schema adherence.
Suppose a smaller, cheaper model costs $0.0020 per attempt, while a frontier model costs $0.0072 per attempt. On paper, the smaller model appears 72% cheaper. But what if the smaller model has a 60% first-pass success rate, requiring automatic retries when JSON validation fails or policy citations are omitted?
That success rate alone doesn't prove the cheaper model costs more per success. Under constant independent 60% success probability, fixed $0.0020 attempt cost, unlimited retries, and no other costs, the expectation is $0.0020 / 0.60 ≈ $0.003333 per success. Real retries may share the same failure cause, change context, hit limits, or be abandoned. Compare accepted outcomes and all attributable spending instead of inferring a winner from one percentage.
When an attempt fails, you pay for both the failed tokens and the subsequent retry. If an agent workflow invokes tools during each attempt, tool execution fees multiply as well. On top of that, abandoned requests burn tokens without producing business value.
Formulate True Cost per Successful Task () by dividing total spend across all attempts, tool calls, and infrastructure by the count of accepted, successful outcomes:
Suppose two users each submit one task. Task A fails its first attempt, then succeeds on retry. Task B fails its first attempt and the user gives up. All three attempts consume identical tokens costing $0.004320 each. Task A's two attempts cost $0.008640, but the service spent $0.012960 in total to deliver exactly one successful task. Omitting Task B hides the real cost of serving this workload.
@dataclass(frozen=True)
class Attempt:
task_id: str
attempt_id: str
usage: Usage
accepted: bool
def __post_init__(self) -> None:
if not nonempty_text(self.task_id) or not nonempty_text(self.attempt_id) or type(self.usage) is not Usage or type(self.accepted) is not bool:
raise ValueError("attempt needs owned IDs, validated usage, and a known acceptance label")
attempts = [
Attempt("A", "A-1", prefix_cached_policy_answer, False),
Attempt("A", "A-2", prefix_cached_policy_answer, True),
Attempt("B", "B-1", prefix_cached_policy_answer, False),
]
assert len({attempt.attempt_id for attempt in attempts}) == len(attempts)
successful_tasks = {attempt.task_id for attempt in attempts if attempt.accepted}
attempt_spend = sum(price_usage(attempt.usage).total_usd for attempt in attempts)
cost_per_success = attempt_spend / len(successful_tasks) if successful_tasks else None
assert attempt_spend == Decimal("0.012960")
assert cost_per_success == Decimal("0.012960")
interactive_daily = total_with_eval - nightly_batch_cost
high_volume_month = (interactive_daily * Decimal("1.10") + nightly_batch_cost) * 30
no_cache_eval_month = (interactive_daily + nightly_cold_cost) * 30
assert dollars(high_volume_month) == Decimal("780.44")
assert dollars(no_cache_eval_month) == Decimal("803.58")
assert high_volume_month > promotion_policy.monthly_budget_usd
assert no_cache_eval_month > promotion_policy.monthly_budget_usd
print(f"attempts={len(attempts)} successful_tasks={len(successful_tasks)}")
print(f"generation_cost_per_success=${cost_per_success:.6f}")
print(f"ten_percent_interactive_growth=${dollars(high_volume_month)}")
print(f"no_cache_nightly_eval_month=${dollars(no_cache_eval_month)}")attempts=3 successful_tasks=1
generation_cost_per_success=$0.012960
ten_percent_interactive_growth=$780.44
no_cache_nightly_eval_month=$803.58The sensitivity checks model two operational shocks without causing real traffic or evictions. A 10% increase in interactive volume raises the forecast to $780.44. Losing all assumed nightly prefix reads raises it to $803.58. Both forecasts exceed $750. In production, include tool calls, embeddings, and failed attempts before declaring a cheaper route economical.
Produce the next policy input
The upcoming lesson builds model gateway routing and dynamic fallbacks. Pass that gateway a formal budget contract rather than embedding pricing heuristics directly into router code. The contract specifies what downstream routes must enforce: the required citation schema, an approved per-generation cost ceiling, and the rate card version.
Treat maximum_generated_answer_usd as a lane-admission ceiling for a single generation call, not a hard bound on cumulative task spend. If a primary call times out and triggers a fallback, total task spend can exceed this per-call cap. Admission contracts must be paired with gateway-level retry limits, per-task spend counters, and circuit breakers.
The largest supplied interactive unit receipt costs $0.004570, under our $0.005000 lane ceiling. This isn't a measured production maximum or a bound on future calls: the earlier incomplete reasoning fixture costs $0.019860. A gateway needs a conservative input/output/cache-write reservation, compatible provider limits, and reconciliation, not an after-the-fact maximum alone.
@dataclass(frozen=True)
class GatewayBudgetContract:
release_id: str
required_answer_schema: str
maximum_generated_answer_usd: Decimal
monthly_forecast_usd: Decimal
rate_card_id: str
rate_card_digest: str
status: str
def __post_init__(self) -> None:
if not all(nonempty_text(value) for value in (
self.release_id, self.required_answer_schema, self.rate_card_id, self.rate_card_digest,
)) or not valid_decimal(self.maximum_generated_answer_usd) or not valid_decimal(self.monthly_forecast_usd):
raise ValueError("contract needs owned identifiers and finite nonnegative Decimal amounts")
if not nonempty_text(self.status) or self.status not in {"PROMOTE_COST_POLICY", "HOLD_RELEASE"}:
raise ValueError("need a known policy verdict")
observed_max_generated_cost = max(
trace_unit_spend(trace)
for trace in traces
if trace.usage is not None and not trace.batch
)
per_answer_budget_usd = Decimal("0.005000")
assert observed_max_generated_cost <= per_answer_budget_usd
status = "PROMOTE_COST_POLICY" if quality_passed and budget_passed else "HOLD_RELEASE"
gateway_contract = GatewayBudgetContract(
release_id=release_id,
required_answer_schema="cited-support-answer-v3",
maximum_generated_answer_usd=per_answer_budget_usd,
monthly_forecast_usd=dollars(monthly_forecast),
rate_card_id=rate_card.rate_card_id,
rate_card_digest=rate_digest(rate_card),
status=status,
)
assert gateway_contract.status == "PROMOTE_COST_POLICY"
assert observed_max_generated_cost == Decimal("0.004570")
assert gateway_contract.maximum_generated_answer_usd == Decimal("0.005000")
assert incomplete_cost.total_usd > gateway_contract.maximum_generated_answer_usd
print(f"status={gateway_contract.status}")
print(f"required_schema={gateway_contract.required_answer_schema}")
print(f"observed_max_generated_answer=${observed_max_generated_cost:.6f}")
print(f"max_generated_answer=${gateway_contract.maximum_generated_answer_usd:.6f}")
print(f"reasoning_fixture_exceeds_lane={incomplete_cost.total_usd > gateway_contract.maximum_generated_answer_usd}")
print("next_step=apply_contract_in_gateway_routing")status=PROMOTE_COST_POLICY
required_schema=cited-support-answer-v3
observed_max_generated_answer=$0.004570
max_generated_answer=$0.005000
reasoning_fixture_exceeds_lane=True
next_step=apply_contract_in_gateway_routingWhat this estimate leaves unresolved
These fixtures demonstrate the core mechanics of an auditable cost ledger: separating input from output, isolating cached reads, checking context boundaries, pinning versioned rate cards, and capturing the retry tax. They don't represent a complete production operating model.
In production, retain real attempt receipts and connect them to OpenTelemetry spans and task outcomes. Reconcile provider statements, negotiated rates, cache writes, tier/region surcharges, rounding, credits, and failed calls. Keep separate columns for embeddings, storage, tool execution, and infrastructure. Trace IDs and evidence labels aren't an immutable audit trail by themselves.
A timeout may leave work running and usage unresolved. Reserve spend before admission, settle it from receipts, and use atomic accounting when calls race. Per-call and per-task limits need retry/fallback policies and explicit exhaustion behavior. This lab emits a symbolic contract and verdict; it enforces no live monthly, task, or call cap. Neither hypothetical savings nor a passing format label funds another generation.