Chapter 8 · Cost, Latency, and Caching Fundamentals
Cost and Latency Fundamentals
- Page 15 of 22
- 25 min read
Mateo in Bogotá ships the EvidenceDesk prototype to his team. A week later the bill is higher than planned and answers feel slow. Nothing is broken: real use brings more requests and retries than testing did. This page teaches the first habits that keep an application affordable and responsive: know what drives cost, record usage, measure time, and skip a call you already paid for. The new code is evidencedesk/usage.py and evidencedesk/cache.py. It runs offline with earlier fake clients and an invented price table, so the numbers show mechanisms, not real prices or speed.
What you will learn
- The four cost drivers, and how to estimate and record cost with exact arithmetic.
- How to measure latency with an injected clock, and first-chunk versus full-response time.
- How to build a deterministic cache key and an in-memory cache that skips repeated calls.
- When a cached answer goes stale, and why private answers are not cached by default.
Identify the main cost drivers
Four things move the bill: input tokens (everything you send, on every call), output tokens (everything the model writes), the number of calls, and retries (a retried call is billed again). To see them, take the RAG request from RAG Ingestion and Citation Flow and weigh its parts with estimate_tokens (characters divided by four, rounded up: a rough size, not a tokeniser).
import json
from pathlib import Path
from evidencedesk.chunking import chunk_corpus
from evidencedesk.context import estimate_tokens
from evidencedesk.rag import RAG_RULES, SAMPLE_REPLIES, Retriever, build_request
chunks = chunk_corpus(Path("kit/corpus"))
retriever = Retriever(chunks, clock=lambda: "2026-10-12T09:00:00Z")
Q = "How long do I have to return a bike?"
found = retriever.retrieve(Q)
request = build_request(Q, found)
print("part".ljust(26), "tokens")
print("system rules".ljust(26), estimate_tokens(RAG_RULES))
for r in found:
print(f"evidence {r.chunk.id}".ljust(26), estimate_tokens(r.chunk.text))
print("whole user message".ljust(26), estimate_tokens(request.messages[1].content), "(with tags)")
reply = estimate_tokens(json.dumps(SAMPLE_REPLIES[Q]))
print("reply (JSON)".ljust(26), reply, f"(the cap is max_tokens={request.max_tokens})")
print()
print("scenario".ljust(30), "calls input output billed")
for label, k, calls in (("k=3, one call", 3, 1), ("k=1, one call", 1, 1), ("k=3, two retries", 3, 3)):
size = sum(estimate_tokens(m.content) for m in build_request(Q, retriever.retrieve(Q, k=k)).messages)
print(f"{label:<30}{calls:>5}{size * calls:>7}{reply * calls:>8}{(size + reply) * calls:>8}")part tokens
system rules 116
evidence kc-returns-v3:c0 58
evidence kc-warranty-v2:c0 41
evidence kc-returns-v3:c1 31
whole user message 210 (with tags)
reply (JSON) 39 (the cap is max_tokens=200)
scenario calls input output billed
k=3, one call 1 326 39 365
k=1, one call 1 215 39 254
k=3, two retries 3 978 117 1095The rules (116 tokens) travel again on every call; the three chunks are only 130 together; the user message is 210 because each chunk is wrapped in labelled tags (Context Assembly and Trust Boundaries). The reply is 39 tokens, far below the cap of 200: max_tokens is a ceiling, not a charge. One evidence chunk instead of three cuts the input by a third, and two retries triple everything, which is why Adapters, Configuration, and Error Handling bounded them.
Estimate cost from a hypothetical price table
Providers quote prices per million tokens, with separate rates for input, output and cached input. Prices change, so our code never holds a real one: NOT_A_REAL_PRICE_TABLE has two invented models and an invented unit, the credit. Rates are Decimal values per 1M tokens, so sums are exact (floats drift: 0.1 + 0.2 is 0.30000000000000004).
# file: kit/evidencedesk/usage.py
"""Cost and latency: a hypothetical price table, a usage ledger, and clocks you can inject."""
from __future__ import annotations
import time
from collections.abc import Callable, Iterable, Iterator
from dataclasses import dataclass
from decimal import Decimal
from evidencedesk.context import estimate_tokens
from evidencedesk.llm import LLMClient, LLMRequest, LLMResponse
from evidencedesk.streaming import Assembled, StreamEvent, assemble
# --- 1. A hypothetical price table ---
PER = Decimal(1_000_000) # rates are "credits per 1M tokens"; a credit is a made-up unit
@dataclass(frozen=True)
class Rate:
input: Decimal
output: Decimal
cached_input: Decimal
NOT_A_REAL_PRICE_TABLE: dict[str, Rate] = { # invented numbers for tests; never a real provider price
"fake-model": Rate(Decimal("3"), Decimal("12"), Decimal("0.75")),
"fake-small": Rate(Decimal("0.6"), Decimal("2.4"), Decimal("0.15")),
}
class UnknownModel(KeyError):
"""No rate for this model: stop, do not count it as free."""Provider note (as of 2026-10-12): the three major providers quote text prices per 1M tokens, input and output separately, per model. Cached input costs less than normal input; one provider adds a cache-write rate, another a storage charge. Reasoning ("thinking") tokens are billed as output. Batch is a cheaper asynchronous tier; some models charge more for very long prompts; tools such as web search can add per-call fees. We print no figures: they change. Check the current documentation.
Record usage metadata when available
Most providers return token counts per call; Provider-Neutral LLM Clients normalised them into Usage. A UsageRecord adds what a ledger needs: a request id, the model, how many input tokens a cache served, an estimated flag and the attempt number. Two rules: input tokens include the cached ones, so cached_tokens can never exceed them (providers report this differently; normalising is the point). And when a response has no usage, we estimate it and mark the record estimated, never zero. A missing price is an error too: UnknownModel stops the run instead of counting a call as free.
# file: kit/evidencedesk/usage.py (append)
# --- 2. Usage records ---
def request_tokens(request: LLMRequest) -> int:
"""Rough input size: ceil(chars / 4) per message."""
return sum(estimate_tokens(m.content) for m in request.messages)
@dataclass(frozen=True)
class UsageRecord:
request_id: str
model: str
input_tokens: int # ALL input tokens, cached ones included
output_tokens: int
cached_tokens: int = 0 # the part of input_tokens a provider cache served
estimated: bool = False # True: no usage from the provider, we guessed
attempt: int = 1 # 1 = first try, 2 and more = a retry
def __post_init__(self) -> None:
if min(self.input_tokens, self.output_tokens, self.cached_tokens) < 0 or self.cached_tokens > self.input_tokens:
raise ValueError("token counts must be >= 0 and cached_tokens <= input_tokens")
if self.attempt < 1:
raise ValueError("attempt starts at 1")
def record_usage(request: LLMRequest, response: LLMResponse, request_id: str, *, cached_tokens: int = 0,
attempt: int = 1) -> UsageRecord:
"""Keep reported usage; if there is none (0 + 0 with text), estimate it and flag it."""
usage = response.usage
if usage.total_tokens == 0 and response.text:
out = min(estimate_tokens(response.text), request.max_tokens)
return UsageRecord(request_id, response.model, request_tokens(request), out, 0, True, attempt)
return UsageRecord(request_id, response.model, usage.input_tokens, usage.output_tokens, cached_tokens, False, attempt)
def cost_of(record: UsageRecord, table: dict[str, Rate] = NOT_A_REAL_PRICE_TABLE) -> Decimal:
"""Exact cost in made-up credits: fresh input + cached input + output."""
try:
rate = table[record.model]
except KeyError:
raise UnknownModel(record.model) from None
fresh = record.input_tokens - record.cached_tokens
return (fresh * rate.input + record.cached_tokens * rate.cached_input + record.output_tokens * rate.output) / PER
@dataclass(frozen=True)
class Totals:
calls: int
retries: int
input_tokens: int
output_tokens: int
cached_tokens: int
estimated: int
cost: Decimal
class Ledger:
"""One record per model call. A cache hit makes no call, so it adds none."""
def __init__(self, table: dict[str, Rate] = NOT_A_REAL_PRICE_TABLE) -> None:
self.table, self.records = table, []
def add(self, record: UsageRecord) -> Decimal:
cost = cost_of(record, self.table) # raises UnknownModel first
self.records.append(record)
return cost
def summary(self) -> Totals:
r = self.records
return Totals(len(r), sum(x.attempt > 1 for x in r), sum(x.input_tokens for x in r), sum(x.output_tokens for x in r),
sum(x.cached_tokens for x in r), sum(x.estimated for x in r), sum((cost_of(x, self.table) for x in r), Decimal(0)))Provider note (as of 2026-10-12): usage names differ. One provider reports a
usageobject withinput_tokens_details.cached_tokens. A second reportsinput_tokens,cache_read_input_tokensandcache_creation_input_tokens, and returns a request id in therequest-idheader. A third reportsusageMetadata(promptTokenCount,cachedContentTokenCount) on one API path and other names on its newer path. Where usage appears inside a stream is not confirmed for every provider here. Check the current documentation.
Reported usage from the FakeClient fixture, an estimate for the hand-written RAG fake (it reports none), 30 cached tokens, and a call retried twice. Fakes are stand-ins: real models are less predictable, which is exactly why you test.
from dataclasses import replace
from evidencedesk.llm import FakeClient, LLMRequest, Message
from evidencedesk.rag import SAMPLE_REPLIES, RecordedReplies
from evidencedesk.usage import NOT_A_REAL_PRICE_TABLE, Ledger, UnknownModel, UsageRecord, cost_of, record_usage
for name, rate in NOT_A_REAL_PRICE_TABLE.items():
print(name, "per 1M tokens: input", rate.input, "| output", rate.output, "| cached input", rate.cached_input)
ask = LLMRequest((Message("user", "How long do I have to return a new bike?"),))
reported = record_usage(ask, FakeClient().complete(ask), "req-001")
rag_request = build_request(Q, found)
silent = record_usage(rag_request, RecordedReplies(SAMPLE_REPLIES).complete(rag_request), "req-002")
cached = UsageRecord("req-003", "fake-model", input_tokens=reported.input_tokens, output_tokens=19, cached_tokens=30)
try:
cost_of(silent)
except UnknownModel as error:
print("no rate for", error)
silent = replace(silent, model="fake-model") # price it as the model it stands in for
for rec in (reported, silent, cached):
print(rec.request_id, rec.input_tokens, rec.output_tokens, rec.cached_tokens, "estimated" if rec.estimated else "reported", cost_of(rec))
ledger = Ledger()
for attempt in (1, 2, 3): # the same call, tried three times
ledger.add(UsageRecord("req-004", "fake-model", 400, 60, attempt=attempt))
print(ledger.summary())fake-model per 1M tokens: input 3 | output 12 | cached input 0.75
fake-small per 1M tokens: input 0.6 | output 2.4 | cached input 0.15
no rate for 'recorded'
req-001 41 19 0 reported 0.000351
req-002 326 39 0 estimated 0.001446
req-003 41 19 30 reported 0.0002835
Totals(calls=3, retries=2, input_tokens=1200, output_tokens=180, cached_tokens=0, estimated=0, cost=Decimal('0.00576'))The fake's model name, 'recorded', has no rate and fails loudly, so we price it as the model it stands in for. req-002 is flagged estimated: we guessed its size from characters, and a ledger should say how many of its rows are guesses. With 30 of 41 input tokens served from a cache, req-003 costs 0.0002835 instead of 0.000351. The last line is the retry driver in numbers: one request, three attempts, retries=2, three times the cost.
Measure latency locally
Latency is how long the caller waits. A Timer reads a clock passed in. The default is time.perf_counter(), a monotonic timer for short intervals; tests pass a ManualClock that moves only when told, so they can assert an exact number. SlowClient advances that clock by a base delay plus a delay per output token: a stand-in for the network, not a measurement of one.
# file: kit/evidencedesk/usage.py (append)
# --- 3. Latency with an injected clock ---
Clock = Callable[[], float]
class ManualClock:
"""Moves only when told. No wall clock."""
def __init__(self, start: float = 0.0) -> None:
self.now = start
def __call__(self) -> float:
return self.now
def advance(self, seconds: float) -> None:
self.now += seconds
class Timer:
"""`with Timer(clock) as t: ...`, then `t.elapsed`. Tests pass a ManualClock."""
def __init__(self, clock: Clock = time.perf_counter) -> None:
self.clock, self.start, self.elapsed = clock, 0.0, 0.0
def __enter__(self) -> Timer:
self.start = self.clock()
return self
def __exit__(self, *exc: object) -> None:
self.elapsed = self.clock() - self.start
class SlowClient:
"""Moves a ManualClock: a base delay plus a delay per output token. A stand-in for the network."""
def __init__(self, inner: LLMClient, clock: ManualClock, base_s: float = 0.4, per_output_token_s: float = 0.02) -> None:
self.inner, self.clock, self.base_s, self.per_token_s = inner, clock, base_s, per_output_token_s
def complete(self, request: LLMRequest) -> LLMResponse:
response = self.inner.complete(request)
self.clock.advance(self.base_s + self.per_token_s * response.usage.output_tokens)
return responsefrom evidencedesk.usage import ManualClock, SlowClient, Timer
clock = ManualClock()
slow = SlowClient(FakeClient(), clock) # 0.4 s + 0.02 s per output token
with Timer(clock) as t:
slow.complete(ask)
print("fake clock:", round(t.elapsed, 3), "s for 19 output tokens")
with Timer() as real: # the default clock is time.perf_counter()
FakeClient().complete(ask)
print("real clock used:", real.elapsed >= 0, "(value not printed)")fake clock: 0.78 s for 19 output tokens
real clock used: True (value not printed)The fake clock gives exactly 0.4 + 0.02 × 19 = 0.78 seconds, every run. The real clock works too, but its value changes every run, so we print only that it ran. Never assert on it: a busy machine would make the test flaky.
Distinguish first-token and full-response latency
A streamed answer has two times. Time to first chunk is how long before the user sees anything; full-response time is how long before the answer is complete. A chat screen feels fast when the first is short, even if the second is long. A non-streamed call has one time only. timed_stream reads a stream (Streaming and Structured Results) and records both; slow_stream delays a fake stream on the manual clock.
# file: kit/evidencedesk/usage.py (append)
# --- 4. First-chunk and full-response latency ---
@dataclass(frozen=True)
class StreamTiming:
first_chunk_s: float | None # None: no chunk arrived
total_s: float
chunks: int
def slow_stream(events: Iterable[StreamEvent], clock: ManualClock, first_s: float = 0.4, per_chunk_s: float = 0.05) -> Iterator[StreamEvent]:
"""Delay a fake stream: first chunk after first_s, each later one per_chunk_s after the last."""
first = True
for event in events:
if event.kind == "chunk":
clock.advance(first_s if first else per_chunk_s)
first = False
yield event
def timed_stream(events: Iterable[StreamEvent], clock: Clock = time.perf_counter) -> tuple[Assembled, StreamTiming]:
"""Read a stream to its end; record time to the first chunk and to the last event."""
start, first, chunks, seen = clock(), None, 0, []
for event in events:
if event.kind == "chunk":
chunks += 1
if first is None:
first = clock() - start
seen.append(event)
return assemble(seen), StreamTiming(first, clock() - start, chunks)from evidencedesk.streaming import FakeStreamClient
from evidencedesk.usage import slow_stream, timed_stream
clock = ManualClock()
events = slow_stream(FakeStreamClient(chunk_size=20).stream(ask), clock, first_s=0.4, per_chunk_s=0.1)
done, timing = timed_stream(events, clock)
print(done.status, "|", done.text)
print(f"chunks={timing.chunks} first chunk after {timing.first_chunk_s:.2f} s whole answer after {timing.total_s:.2f} s")complete | You can return a new, unridden bike within 30 days of delivery for a full refund.
chunks=5 first chunk after 0.40 s whole answer after 0.80 sThe user waits 0.40 s for the first words and 0.80 s for the whole answer. A logged total alone cannot tell a slow start from a slow finish, and they have different fixes (a smaller prompt, a shorter answer).
Build a deterministic cache key
The cheapest call is the one you do not make. A cache stores an answer under a key, which must contain everything that can change the answer and nothing that varies for no reason. Ours is a SHA-256 hash (first 16 hex digits) of canonical JSON: model and max_tokens, the messages with whitespace collapsed (case is kept: RD2-24 is not rd2-24), a prompt_version, the evidence as chunk-id@version, a corpus_version and a scope. The messages already hold the rules and evidence text; the labels also cover changes invisible there (new post-processing code) and name what to purge.
Not in the key: request ids, timestamps or anything random; every key would be new and the cache would never hit. In the key whenever the answer depends on who asks: the user or tenant id, as the scope.
# file: kit/evidencedesk/cache.py
"""Exact-match response cache: a deterministic key, a policy per request class, a TTL on an injected clock."""
from __future__ import annotations
import hashlib
import json
import re
import time
from collections.abc import Callable, Iterable
from dataclasses import dataclass
from evidencedesk.llm import LLMClient, LLMRequest, LLMResponse
from evidencedesk.rag import Retrieved
# --- 1. The key ---
def normalise(text: str) -> str:
"""Collapse whitespace; keep case (RD2-24 is not rd2-24)."""
return re.sub(r"\s+", " ", text).strip()
def evidence_refs(retrieved: Iterable[Retrieved]) -> tuple[str, ...]:
"""The evidence set as chunk-id@version."""
return tuple(f"{r.chunk.id}@{r.chunk.version}" for r in retrieved)
def cache_key(request: LLMRequest, *, prompt_version: str, evidence: Iterable[str] = (), corpus_version: str = "",
scope: str = "") -> str:
"""Hash of everything that can change the answer. Nothing random in it: no ids, no times."""
material = {
"model": request.model, "max_tokens": request.max_tokens,
"messages": [[m.role, normalise(m.content)] for m in request.messages],
"prompt_version": prompt_version, "evidence": sorted(evidence), "corpus_version": corpus_version, "scope": scope,
}
text = json.dumps(material, sort_keys=True, ensure_ascii=False, separators=(",", ":"))
return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16]from evidencedesk.cache import cache_key, evidence_refs
base = dict(prompt_version="v1", evidence=evidence_refs(found), corpus_version="2026-10-12")
rag_a = build_request(Q, found)
variants = {
"same request again": (rag_a, {}),
"extra spaces in the question": (build_request(" How long do I have to return a bike? ", found), {}),
"different model": (LLMRequest(rag_a.messages, model="fake-small"), {}),
"different max_tokens": (LLMRequest(rag_a.messages, max_tokens=50), {}),
"different letter case": (build_request(Q.lower(), found), {}),
"prompt_version v2": (rag_a, {"prompt_version": "v2"}),
"one evidence chunk fewer": (rag_a, {"evidence": evidence_refs(found[:-1])}),
"corpus_version changed": (rag_a, {"corpus_version": "2026-10-13"}),
"scope user:nadia": (rag_a, {"scope": "user:nadia"}),
}
key0 = cache_key(rag_a, **base)
print("base key:", key0[:8], "...")
for label, (req, change) in variants.items():
print(f"{label:<30}", "same key" if cache_key(req, **{**base, **change}) == key0 else "NEW key")base key: 81f3f88d ...
same request again same key
extra spaces in the question same key
different model NEW key
different max_tokens NEW key
different letter case NEW key
prompt_version v2 NEW key
one evidence chunk fewer NEW key
corpus_version changed NEW key
scope user:nadia NEW keyExtra spaces do not matter; every other change does, including letter case and max_tokens (a shorter cap can cut the answer). The key is only as good as its list of inputs: a setting that changes the answer but is missing from it returns wrong answers fast. A cache also freezes the first answer it saw, which matters when sampling is random (temperature 0 does not guarantee identical output either: test with fixtures, not hope).
Implement an in-memory response cache
ResponseCache is a dictionary with a time limit (TTL) on an injected clock, a size limit, hit and miss counters, and tags on each entry (for invalidation below). CachedClient wraps any client: it builds the key, answers from the cache on a hit, and calls the model only on a miss. A cut-off answer (stop_reason not end) is never stored. The policy is explained in the privacy section; here FAQ_ONLY allows the class faq.
# file: kit/evidencedesk/cache.py (append)
# --- 2. The store ---
@dataclass
class _Entry:
response: LLMResponse
stored_at: float
tags: frozenset[str]
class ResponseCache:
"""In memory. Entries expire after ttl_s on the injected clock; the oldest goes first when full."""
def __init__(self, ttl_s: float = 300.0, clock: Callable[[], float] = time.monotonic, max_entries: int = 256) -> None:
self.ttl_s, self.clock, self.max_entries = ttl_s, clock, max_entries
self.entries: dict[str, _Entry] = {}
self.hits = self.misses = self.expired = 0
def __len__(self) -> int:
return len(self.entries)
def get(self, key: str) -> LLMResponse | None:
entry = self.entries.get(key)
if entry is not None and self.clock() - entry.stored_at >= self.ttl_s:
del self.entries[key]
self.expired += 1
entry = None
if entry is None:
self.misses += 1
return None
self.hits += 1
return entry.response
def put(self, key: str, response: LLMResponse, tags: Iterable[str] = ()) -> None:
if key not in self.entries and len(self.entries) >= self.max_entries:
del self.entries[next(iter(self.entries))]
self.entries[key] = _Entry(response, self.clock(), frozenset(tags))
def invalidate(self, tag: str) -> int:
"""Remove entries with this tag; return how many."""
doomed = [k for k, e in self.entries.items() if tag in e.tags]
for key in doomed:
del self.entries[key]
return len(doomed)
def clear(self) -> None:
self.entries.clear()# file: kit/evidencedesk/cache.py (append)
# --- 3. The policy and the client ---
@dataclass(frozen=True)
class CachePolicy:
"""Which request classes may be cached. The default: none."""
shared: frozenset[str] = frozenset() # any user may get the cached answer
per_user: frozenset[str] = frozenset() # cached inside one user's scope only
NO_CACHE = CachePolicy()
FAQ_ONLY = CachePolicy(shared=frozenset({"faq"}))
@dataclass(frozen=True)
class Cached:
response: LLMResponse
status: str # hit | miss | bypass
key: str | None = None # None for bypass
class CachedClient:
"""`ask` says what happened (hit, miss, bypass); `complete` keeps the LLMClient shape."""
def __init__(self, inner: LLMClient, cache: ResponseCache, policy: CachePolicy = NO_CACHE, prompt_version: str = "v1") -> None:
self.inner, self.cache, self.policy, self.prompt_version = inner, cache, policy, prompt_version
self.bypassed = 0
def ask(self, request: LLMRequest, *, request_class: str = "unclassified", user_id: str | None = None,
evidence: Iterable[str] = (), corpus_version: str = "") -> Cached:
if request_class in self.policy.shared:
scope = ""
elif request_class in self.policy.per_user and user_id:
scope = f"user:{user_id}"
else: # not allowed, or no user to isolate by: fail closed
self.bypassed += 1
return Cached(self.inner.complete(request), "bypass")
refs = tuple(evidence)
key = cache_key(request, prompt_version=self.prompt_version, evidence=refs, corpus_version=corpus_version, scope=scope)
found = self.cache.get(key)
if found is not None:
return Cached(found, "hit", key)
response = self.inner.complete(request)
if response.stop_reason == "end": # a cut-off answer is never cached
tags = {f"prompt:{self.prompt_version}", f"model:{request.model}", f"corpus:{corpus_version}"}
tags |= {"doc:" + ref.split(":")[0] for ref in refs}
self.cache.put(key, response, tags)
return Cached(response, "miss", key)
def complete(self, request: LLMRequest, **kwargs: object) -> LLMResponse:
return self.ask(request, **kwargs).response # type: ignore[arg-type]Five identical questions, without and with the FAQ class cached. A hit makes no model call, so it adds no ledger record.
from evidencedesk.cache import FAQ_ONLY, NO_CACHE, CachedClient, ResponseCache
from evidencedesk.usage import Ledger
def run(policy, n=5):
clock, inner = ManualClock(), FakeClient()
client, ledger = CachedClient(inner, ResponseCache(300, clock), policy), Ledger()
for i in range(n):
got = client.ask(ask, request_class="faq")
if got.status != "hit": # a hit makes no model call, so there is nothing to record
ledger.add(record_usage(ask, got.response, f"req-{i}"))
return client, inner, ledger.summary()
for label, policy in (("no cache", NO_CACHE), ("faq cached", FAQ_ONLY)):
client, inner, total = run(policy)
print(f"{label:<11} model calls={len(inner.calls)} tokens={total.input_tokens + total.output_tokens} cost={total.cost}"
f" hits={client.cache.hits} misses={client.cache.misses} bypassed={client.bypassed}")no cache model calls=5 tokens=300 cost=0.001755 hits=0 misses=0 bypassed=5
faq cached model calls=1 tokens=60 cost=0.000351 hits=4 misses=1 bypassed=0One model call instead of five: 60 tokens instead of 300, 0.000351 credits instead of 0.001755, with 4 hits and 1 miss. This shows the mechanism; five calls to a fake are not a benchmark. A real saving depends on how often users repeat a question, which you learn only by counting (the next page does).
Define cache invalidation rules
A cached answer is valid only while its sources are unchanged. Our rules: invalidate when the prompt version, the model or its settings, the evidence set (a chunk id or version) or the corpus version changes, and when the TTL runs out. All of these are in the key, so a change makes the old entry unreachable: the new request simply misses. The old entry still uses memory, so tags (doc:, prompt:, model:, corpus:) let you purge by name, which also serves a "delete this document now" request. The evidence carries the chunk versions of RAG Ingestion and Citation Flow and RAG Reliability and Evaluation Data; the corpus holds two returns pages that disagree, and a stale answer would quote the wrong number of days.
from dataclasses import replace
clock, inner = ManualClock(), FakeClient()
client = CachedClient(inner, ResponseCache(3600, clock), FAQ_ONLY, prompt_version="v1")
refs = evidence_refs(found)
print("chunk versions:", [ref.split("@")[1] for ref in refs], "->", [client.ask(ask, request_class="faq", evidence=refs).status for _ in range(2)])
newer = [replace(r.chunk, version=r.chunk.version + 1) for r in found] # the returns page was updated
refs_v2 = tuple(f"{c.id}@{c.version}" for c in newer)
print("evidence changed ->", client.ask(ask, request_class="faq", evidence=refs_v2).status)
client.prompt_version = "v2"
print("prompt version changed ->", client.ask(ask, request_class="faq", evidence=refs).status)
print("entries before purge:", len(client.cache), "| removed by tag doc:kc-returns-v3:", client.cache.invalidate("doc:kc-returns-v3"),
"| left:", len(client.cache))
client.prompt_version = "v1"
print("old request after purge ->", client.ask(ask, request_class="faq", evidence=refs).status)chunk versions: ['3', '2', '3'] -> ['miss', 'hit']
evidence changed -> miss
prompt version changed -> miss
entries before purge: 3 | removed by tag doc:kc-returns-v3: 3 | left: 0
old request after purge -> missThe same question hits twice; a newer chunk version (even with the same text) and a new prompt version each miss. Three entries now exist. One purge by tag removes all three, so the original request misses afterwards.
Avoid caching sensitive responses by default
A cache is a second copy of your users' data. So the default is to cache nothing (NO_CACHE); a request class is cached only if the policy names it. shared classes (public FAQ answers) may be served to anyone. per_user classes are cached only inside one user's scope, and a request with no user id is not cached (fail closed). Everything else, such as account data, bypasses the cache. The TTL keeps copies short-lived.
from evidencedesk.cache import CachePolicy
policy = CachePolicy(shared=frozenset({"faq"}), per_user=frozenset({"orders"}))
clock, inner = ManualClock(), FakeClient()
client = CachedClient(inner, ResponseCache(60, clock), policy)
steps = [
("faq, nadia", dict(request_class="faq", user_id="nadia")),
("faq, omar (shared)", dict(request_class="faq", user_id="omar")),
("orders, nadia", dict(request_class="orders", user_id="nadia")),
("orders, nadia again", dict(request_class="orders", user_id="nadia")),
("orders, omar", dict(request_class="orders", user_id="omar")),
("orders, no user", dict(request_class="orders")),
("account (not allowed)", dict(request_class="account", user_id="nadia")),
]
for label, kw in steps:
print(f"{label:<24}{client.ask(ask, **kw).status}")
clock.advance(61)
print(f"{'faq after 61 s (ttl 60)':<24}{client.ask(ask, request_class='faq').status}")
print("model calls:", len(inner.calls), "| entries:", len(client.cache))faq, nadia miss
faq, omar (shared) hit
orders, nadia miss
orders, nadia again hit
orders, omar miss
orders, no user bypass
account (not allowed) bypass
faq after 61 s (ttl 60) miss
model calls: 6 | entries: 3Omar gets Nadia's cached FAQ answer (public text) but never her order answer: his order request is a miss of its own. A request with no user, and a class not on the list, bypass the cache. After 61 seconds on the fake clock the 60-second entry has expired and the next request misses.
Provider note (as of 2026-10-12): providers also cache on their side, separately from ours. Prompt caching needs an identical prompt prefix; hits are reported as cached tokens, billed at the lower rate; one provider says its caches are isolated between organisations. Retention differs: one provider deletes inputs and outputs within 30 days (with exceptions); another keeps abuse-monitoring logs up to 30 days and keeps no cached data after 24 hours. On one provider's unpaid tier, content may be used to improve products and read by human reviewers, so send no sensitive or personal data; its retention period was not stated on the pages checked. Check the current documentation and your own agreement.
Exercise: compare cached and uncached calls
The five-question run showed the call-count reduction; now pin it with tests (24 in kit/tests/test_usage.py and kit/tests/test_cache.py). One parametrised test changes a single input (model, max_tokens, letter case, prompt version, evidence, corpus version, scope) and asserts a different key; another asserts that three identical calls make one model call. sed removes the wall-clock seconds.
python -m pytest -q kit/tests/test_usage.py kit/tests/test_cache.py | sed 's/ in [0-9.]*s//'........................ [100%]
24 passedPassing tests show the keys separate for those seven inputs, not that the list is complete. Honest limits: this cache matches exact requests, so a reworded question is a new key; matching similar questions by embedding (semantic caching) is a different, riskier design that can answer the wrong question, and we do not build it.
Where this shows up at work
Teams add a usage ledger first: a bill without a breakdown ("which feature, which model, how many retries?") cannot be acted on. Product teams watch first-chunk time because users feel it. A FAQ bot is the classic cache candidate; an account assistant is the classic leak when the key lacks the user. Reviewers ask of every cache: what is stored, for how long, who can read it?
Common mistakes
- A key without the user or tenant id. One user receives another's answer. Fix: put a scope in the key, or do not cache that class.
- Caching by question text only. A new prompt, model or document version keeps serving the old answer. Fix: put them in the key.
- Treating missing usage as zero. The ledger under-reports silently. Fix: estimate and flag
estimated. - Asserting on real elapsed time. Tests flake. Fix: inject a clock.
- Floats for money, or real prices in code. Drift and stale numbers. Fix:
Decimaland an invented table.
Try it yourself
- Easy. Send the three questions of Provider-Neutral LLM Clients through
FakeClientwith the modelfake-small, add a record for each to aLedgerand print the summary. - Medium. Take the interrupted stream of Streaming and Structured Results (
stop_after=2). Print its status, usage and timing, then write an estimatedUsageRecordfor it and price it. - Hard. Cache a RAG answer from RAG Reliability and Evaluation Data: ask the same question twice, then rebuild the retriever with the returns chunks at a newer version and ask again. Count the model calls.
Answers
Easy.
from evidencedesk.usage import Ledger, record_usage
small, ledger = FakeClient(), Ledger()
for n, question in enumerate(small.fixtures, 1):
req = LLMRequest((Message("user", question),), model="fake-small")
resp = small.complete(req)
ledger.add(record_usage(req, resp, f"e-{n}"))
print(question[:32].ljust(34), resp.stop_reason.ljust(7), resp.usage.total_tokens)
print(ledger.summary())How long do I have to return a n end 60
How long does standard delivery end 58
How should I store the e-bike ba length 65
Totals(calls=3, retries=0, input_tokens=124, output_tokens=59, cached_tokens=0, estimated=0, cost=Decimal('0.0002160'))Medium. A dropped stream sends no done event, so no usage: estimate and flag it.
from evidencedesk.context import estimate_tokens
from evidencedesk.streaming import FakeStreamClient
from evidencedesk.usage import request_tokens, slow_stream, timed_stream
clock = ManualClock()
events = slow_stream(FakeStreamClient(chunk_size=20, stop_after=2).stream(ask), clock)
done, timing = timed_stream(events, clock)
print(done.status, "| usage:", done.usage, "|", timing)
rec = UsageRecord("e-stream", "fake-model", request_tokens(ask), estimate_tokens(done.text), estimated=True)
print("estimated:", rec.estimated, "| cost:", cost_of(rec))interrupted | usage: None | StreamTiming(first_chunk_s=0.4, total_s=0.45, chunks=2)
estimated: True | cost: 0.00015Hard. The cache key carries the evidence versions, so a version bump misses even though the text is identical.
from dataclasses import replace
from evidencedesk.cache import FAQ_ONLY, CachedClient, ResponseCache, evidence_refs
from evidencedesk.rag import SAMPLE_REPLIES, RecordedReplies, Retriever, build_request
def cached_rag(client, retriever, question):
found = retriever.retrieve(question)
return client.ask(build_request(question, found), request_class="faq", evidence=evidence_refs(found))
replies = RecordedReplies(SAMPLE_REPLIES)
client = CachedClient(replies, ResponseCache(3600, ManualClock()), FAQ_ONLY)
print([cached_rag(client, retriever, Q).status for _ in range(2)], "model calls:", len(replies.calls))
bumped = [replace(c, version=c.version + 1) if c.doc_id == "kc-returns-v3" else c for c in chunks]
print(cached_rag(client, Retriever(bumped, clock=lambda: "2026-10-12T09:00:00Z"), Q).status, "model calls:", len(replies.calls))['miss', 'hit'] model calls: 1
miss model calls: 2Pass checklist:
- Easy: three calls, no retries, nothing estimated; the third ends with
lengthand is still billed, atfake-smallrates. - Medium: status
interrupted, usageNone, first chunk 0.4 s; the record isestimated: a guess, not a bill. - Hard:
['miss', 'hit']with one model call, then amissand two calls after the version bump.
Summary
- Cost comes from input tokens, output tokens, calls and retries; measure each before you optimise.
- Price with an invented table and
Decimal; record usage per call and flag estimates. - Inject the clock to measure latency; keep first-chunk and full-response time apart; never assert on real time.
- A cache key holds everything that can change the answer (model settings, prompt, evidence versions, corpus, user scope) and nothing random.
- Cache nothing by default, isolate by user, expire by TTL, and purge by tag; exact-match caching is the safe starting point.
Next: Budgets, Caching, and Usage Limits uses this ledger and cache to set input and output budgets, bound retries, enforce a daily usage limit, track the cache hit rate and write an honest cost and latency summary.