Cost Optimization for LLM Applications- Caching, Routing, Batching, and Model Selection

Caching, routing, batching and model selection, with Pulse, a multi-tenant social media platform, as the running example

LLM cost optimisation is not choosing the cheapest model. The objective is the lowest total cost per successfully completed workflow at the required quality, safety, privacy, latency and reliability. Four controls do most of the work: cache only reusable work, route by task and risk, batch work that can wait, and select the cheapest model that passes task-specific evaluation. Each is shown below with tested code and then assembled into a production reference design.

Pricing note: all prices are illustrative. Prices, cache rules, batch discounts, model identifiers and context limits change often; load them from reviewed configuration and verify them against current provider documentation before deployment.

1. Pulse: the running system

Pulse offers post summaries and topic tags, comment moderation, personalised recommendations, weekly creator reports and AI-assisted reply suggestions. It serves many creators and organisations, so tenant isolation is mandatory. Posts and comments are untrusted input, and moderation decisions affect reach, reputation and account access, so an LLM score is never a final policy decision.

Measure cost per successful workflow, not cost per API call, and set separate latency objectives for interactive and asynchronous work.

Track quality next to spend: moderation false-positive and false-negative rates, creator edits, appeals and human overrides.

Enforce per-tenant quotas and global budgets, reserve capacity for safety workflows, and never trade safety, privacy or tenant isolation for a higher cache-hit rate.

2. Cost Accounting That Matches the Invoice

Providers report token usage differently, and a cost model that ignores this breaks on real traffic. Anthropic's input_tokens excludes cached tokens: a request that reads 100,000 tokens from cache and sends 50 new ones reports input_tokens: 50. OpenAI's input_tokens includes them. A validation rule such as “cached tokens cannot exceed input tokens” therefore rejects every cached Anthropic call. Normalise usage into four disjoint buckets at the provider boundary, then price them with Decimal.

from collections.abc import Mapping
from dataclasses import dataclass
from decimal import Decimal

MILLION = Decimal(1_000_000)

@dataclass(frozen=True)
class Price:  # USD per million tokens, loaded from reviewed config
    input: Decimal
    cache_write: Decimal  # some providers bill cache writes at a premium
    cache_read: Decimal
    output: Decimal       # reasoning tokens are billed as output

@dataclass(frozen=True)
class Usage:  # provider-neutral; the four buckets never overlap
    uncached_input: int
    cache_write: int
    cache_read: int
    output: int

    def __post_init__(self) -> None:
        for name, value in vars(self).items():
            if type(value) is not int or value < 0:
                raise ValueError(f"{name} must be a non-negative int")

def usage_from_anthropic(u) -> Usage:
    # Anthropic: input_tokens EXCLUDES cache reads and cache writes.
    return Usage(u.input_tokens, u.cache_creation_input_tokens or 0,
                 u.cache_read_input_tokens or 0, u.output_tokens)

def usage_from_openai(u) -> Usage:
    # OpenAI Responses API: input_tokens INCLUDES cached (and cache-write) tokens.
    d = u.input_tokens_details
    read = d.cached_tokens or 0
    write = getattr(d, "cache_write_tokens", 0) or 0  # reported by newer models only
    if read + write > u.input_tokens:
        raise ValueError("inconsistent usage payload")
    return Usage(u.input_tokens - read - write, write, read, u.output_tokens)

PRICES: Mapping[str, Price] = {  # ILLUSTRATIVE ONLY: keyed by logical tier
    "small-fast": Price(Decimal("0.20"), Decimal("0.25"), Decimal("0.02"), Decimal("0.80")),
    "standard": Price(Decimal("2.00"), Decimal("2.50"), Decimal("0.20"), Decimal("8.00")),
    "large-reasoning": Price(Decimal("10.00"), Decimal("12.50"), Decimal("1.00"), Decimal("40.00")),
}

def calculate_cost(model: str, usage: Usage) -> Decimal:
    try:
        p = PRICES[model]
    except KeyError:
        raise ValueError(f"unconfigured model: {model}") from None  # fail closed
    total = (usage.uncached_input * p.input + usage.cache_write * p.cache_write
             + usage.cache_read * p.cache_read + usage.output * p.output)
    return (total / MILLION).quantize(Decimal("0.00000001"))  # matches NUMERIC(18,8)

Cache writes are billed separately: Anthropic charges 1.25× base input for the 5-minute cache and 2× for the 1-hour cache, and newer OpenAI models also bill cache writes. A three-price model (input, cached, output) cannot represent this.

Reasoning models bill hidden reasoning tokens as output, and output is usually priced several times higher than input. Cap them (max_output_tokens, reasoning effort) and include them in budgets.

Use Decimal quantised to the database's NUMERIC(18,8); summing floats over millions of rows drifts. Model batch discounts and long-context price tiers where your provider has them.

3. Caching

What to cache. Reuse a response only when the same request under the same effective configuration is safe to reuse. The key must cover tenant scope, task, input digest, prompt, policy and model versions, and every output-affecting parameter. Never cache authorisation decisions or share user-specific recommendations across users, and avoid shared semantic caching for moderation, enforcement or crisis-related content.

Provider prompt caching is a different mechanism. It discounts repeated prompt prefixes, not whole responses. OpenAI applies it automatically above a minimum prefix length; a stable prompt_cache_key helps requests that share a prefix reach the same cache and, on newer models, isolates cache accounting per customer. Anthropic requires an explicit opt-in with cache_control (minimum 512 to 4,096 tokens depending on the model; 5-minute default TTL, 1 hour optional). Either way, put stable content (system prompt, tool definitions, policy text) first and variable data last, and confirm hits from the usage fields instead of assuming them.

Response cache. This version is async (the worker awaits it), wraps values in a schema-versioned envelope, and implements tenant-wide invalidation with a generation counter.

import hashlib
import json
from typing import Any
from redis.asyncio import Redis

CACHE_MISS = object()
SCHEMA = 1

class TenantResponseCache:
    def __init__(self, redis: Redis, ttl_seconds: int = 900) -> None:
        if ttl_seconds <= 0:
            raise ValueError("ttl_seconds must be positive")
        self.redis, self.ttl = redis, ttl_seconds

    async def make_key(self, *, tenant_id: str, task: str, model: str, prompt_version: str,
                       policy_version: str, parameters: dict[str, Any], input_text: str) -> str:
        if not all([tenant_id, task, model, prompt_version, policy_version]):
            raise ValueError("cache scope fields are required")
        payload = {"task": task, "model": model, "prompt": prompt_version,
                   "policy": policy_version, "params": parameters,
                   "input": hashlib.sha256(input_text.encode()).hexdigest()}
        digest = hashlib.sha256(
            json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
        gen = int(await self.redis.get(f"pulse:cache-gen:{tenant_id}") or 0)
        return f"pulse:llm:v{SCHEMA}:{tenant_id}:g{gen}:{digest}"

    async def get(self, key: str) -> Any:
        raw = await self.redis.get(key)
        if raw is None:
            return CACHE_MISS
        try:
            envelope = json.loads(raw)
            if envelope["schema"] != SCHEMA:
                raise ValueError("schema mismatch")
            return envelope["value"]
        except (ValueError, KeyError, TypeError):
            await self.redis.delete(key)  # corrupt or old entries are misses, never output
            return CACHE_MISS

    async def set(self, key: str, value: dict[str, Any]) -> None:
        envelope = {"schema": SCHEMA, "value": value}
        await self.redis.set(key, json.dumps(envelope, separators=(",", ":"),
                                             ensure_ascii=False), ex=self.ttl)

    async def invalidate_tenant(self, tenant_id: str) -> None:
        # O(1): bump the generation; old keys become unreachable and expire by TTL.
        await self.redis.incr(f"pulse:cache-gen:{tenant_id}")

invalidate_tenant is O(1): bumping the generation makes old keys unreachable and the TTL reclaims them. Never run KEYS, and avoid SCAN sweeps, on production Redis.

The explicit CACHE_MISS sentinel lets valid falsy results such as {"tags": []} be cached. Look up the cache only after authorisation and policy checks (see the worker in section 7).

Use a shared cache across instances, with TTLs set by data volatility, bounded memory (maxmemory plus an eviction policy) and stampede protection (a short SET NX lock) for hot keys. Record hits, misses and token savings without logging raw content.

4. Model Routing

Routing is a governed policy, not a keyword heuristic. It makes two separate decisions: the risk disposition (process, human review or reject) and the model tier for an allowed task. A larger model may reason better, but it is not a safety control. Task types are defined once as a Literal shared by the API and the router; if the two keep separate lists, a topic_tags job accepted by the API silently falls through to human review.

from collections.abc import Mapping
from dataclasses import dataclass
from typing import Literal

TaskType = Literal["topic_tags", "post_summary", "reply_draft", "weekly_report"]  # shared with API
Tier = Literal["small-fast", "standard", "large-reasoning"]
Disposition = Literal["process", "human_review", "reject"]

HIGH_IMPACT = frozenset({"credible_threat", "self_harm", "minor_safety",
                         "doxxing", "targeted_harassment"})
TASK_TIERS: Mapping[str, Tier] = {"topic_tags": "small-fast", "post_summary": "standard",
                                  "reply_draft": "standard", "weekly_report": "standard"}

@dataclass(frozen=True)
class Route:
    disposition: Disposition
    tier: Tier | None
    reason: str

def choose_route(*, task: str, risk: str, uncertainty: float, input_tokens: int,
                 max_input_tokens: Mapping[str, int], review_threshold: float) -> Route:
    if input_tokens <= 0:
        return Route("reject", None, "empty_input")
    if risk in HIGH_IMPACT:
        return Route("human_review", None, "high_impact_policy_case")
    if not 0.0 <= uncertainty <= 1.0:  # also catches NaN: fail closed
        return Route("human_review", None, "invalid_uncertainty")
    if uncertainty >= review_threshold:  # threshold calibrated on labelled data
        return Route("human_review", None, "uncertain_classification")
    tier = TASK_TIERS.get(task)
    if tier is None:
        return Route("human_review", None, "unmapped_task")
    if input_tokens > max_input_tokens[tier]:
        return Route("reject", None, "input_exceeds_context")  # chunk upstream
    return Route("process", tier, "task_policy")

Risk and uncertainty come from a separately evaluated policy classifier, never from caller-supplied fields. Out-of-range values, including NaN, fail closed to review. Calibrate the threshold on labelled data per task and language, measuring false negatives and false positives separately.

A cascade often saves more than static routing: try the small tier, validate the output, and escalate once to the next tier only on failure. Track the escalation rate; a high rate means the small tier is the wrong default.

Check context limits, regional availability, data-processing terms and structured-output support per tier, and pair every route with timeouts, bounded retries with jitter, a circuit breaker and an evaluated fallback.

5. Batching

Batching suits independent, delay-tolerant work such as post backfills, nightly reports, offline embeddings and evaluation runs, never interactive replies. Provider batch APIs (OpenAI Batch API, Anthropic Message Batches) typically discount about 50% in exchange for a completion window of up to 24 hours, but formats, limits and eligibility vary, so read discounts from configuration. The worker below separates transient from permanent failures and caps attempts: retrying a malformed input three times only triples its cost.

from collections.abc import Callable
from dataclasses import dataclass, field
from typing import Any, Literal

ItemStatus = Literal["queued", "completed", "failed_transient", "failed_permanent"]

class TransientError(Exception):
    """Timeout, 429 or 5xx from the provider: safe to retry."""

@dataclass
class BatchItem:
    custom_id: str
    payload: dict[str, Any]
    status: ItemStatus = "queued"
    attempts: int = 0
    result: dict[str, Any] | None = None
    error_code: str | None = None

@dataclass
class BatchJob:
    job_id: str
    items: list[BatchItem] = field(default_factory=list)
    status: str = "queued"

def process_batch(job: BatchJob, process_one: Callable[[dict[str, Any]], dict[str, Any]],
                  max_attempts: int = 3) -> BatchJob:
    ids = [i.custom_id for i in job.items]
    if len(ids) != len(set(ids)):
        raise ValueError("custom_id values must be unique within a batch")
    job.status = "running"
    for item in job.items:
        if item.status in ("completed", "failed_permanent") or item.attempts >= max_attempts:
            continue  # resume safely: never redo finished or exhausted items
        item.attempts += 1
        try:
            item.result = process_one(item.payload)
            item.status, item.error_code = "completed", None
        except TransientError:
            item.status, item.error_code = "failed_transient", "TRANSIENT"
        except Exception as exc:  # bad input or invalid output: retrying won't help
            item.status, item.error_code = "failed_permanent", type(exc).__name__
    done = sum(i.status == "completed" for i in job.items)
    if any(i.status == "failed_transient" and i.attempts < max_attempts for i in job.items):
        job.status = "retry_pending"  # scheduler re-runs after a backoff delay
    elif done == len(job.items):
        job.status = "completed"
    else:
        job.status = "completed_with_errors" if done else "failed"
    return job

A real worker persists each transition transactionally, leases items so two workers never process the same one, respects provider rate limits and stores provider request IDs and usage for reconciliation.

Workflow: validate and deduplicate items (tenant, operation, source version, idempotency key) → persist the parent job and items → submit bounded chunks, keeping provider batch IDs and input checksums → reconcile each item, including malformed outputs → mark the parent terminal only when every item is terminal.

6. Model Selection

Select the cheapest model that meets a measured, task-specific quality bar, benchmarked on representative data including long-tail, multilingual, adversarial and safety-sensitive examples. Compare quality, latency, availability, data residency, structured-output reliability and total cost per successful workflow, which includes retries, escalations and human review. Before dispatch, count tokens with the model's tokenizer or the provider's token-counting endpoint; word-count estimates are for capacity planning only, and billing uses provider-reported usage.

Cut tokens before cutting model size: trim prompts and retrieved context, cap output length, ask for compact structured output, and keep stable prefixes first so prompt caching applies.

Replace generation where you can: embeddings or a fine-tuned small classifier for tagging, deterministic code for extraction.

7. Production Reference Architecture

Request handling, model execution and publishing are separate. The API authenticates the caller, authorises access to the tenant-owned resource, validates input, applies quotas and writes an idempotent job plus an outbox event in one transaction. A worker re-checks authorisation, applies policy and routing, consults the cache, reserves budget, calls the provider, records usage, validates output and persists the result. Publishing is a separate, creator-approved action: LLM output never authorises publishing, deletion, suspension or any enforcement.

Data model (PostgreSQL)

The schema stores everything the API and worker use, including the input text and a request hash for idempotency checks, plus disjoint token buckets matching section 2, an outbox table and an index on the usage foreign key (PostgreSQL does not index foreign keys automatically).

CREATE TABLE llm_jobs (
  job_id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  tenant_id       UUID NOT NULL,
  actor_id        UUID NOT NULL,
  task_type       TEXT NOT NULL,
  source_id       TEXT NOT NULL,
  source_version  TEXT NOT NULL,
  input_text      TEXT NOT NULL,  -- or a pointer to encrypted object storage
  idempotency_key TEXT NOT NULL,
  request_hash    TEXT NOT NULL,  -- detects key reuse with a different body
  status          TEXT NOT NULL CHECK (status IN
                    ('queued','running','needs_review','completed','failed','rejected')),
  prompt_version  TEXT NOT NULL,
  policy_version  TEXT NOT NULL,
  model_name      TEXT,
  result_json     JSONB,
  error_code      TEXT,
  created_at      TIMESTAMPTZ NOT NULL DEFAULT now(),
  updated_at      TIMESTAMPTZ NOT NULL DEFAULT now(),  -- maintained by trigger
  UNIQUE (tenant_id, idempotency_key)
);
CREATE INDEX llm_jobs_tenant_created ON llm_jobs (tenant_id, created_at DESC);

CREATE TABLE job_outbox (
  event_id     BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  job_id       UUID NOT NULL REFERENCES llm_jobs (job_id),
  published_at TIMESTAMPTZ
);
CREATE INDEX job_outbox_pending ON job_outbox (event_id) WHERE published_at IS NULL;

CREATE TABLE llm_usage (
  usage_id              BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  job_id                UUID NOT NULL REFERENCES llm_jobs (job_id),
  provider_request_id   TEXT,
  model_name            TEXT NOT NULL,
  uncached_input_tokens BIGINT NOT NULL CHECK (uncached_input_tokens >= 0),
  cache_write_tokens    BIGINT NOT NULL CHECK (cache_write_tokens >= 0),
  cache_read_tokens     BIGINT NOT NULL CHECK (cache_read_tokens >= 0),
  output_tokens         BIGINT NOT NULL CHECK (output_tokens >= 0),
  cost_usd              NUMERIC(18,8) NOT NULL CHECK (cost_usd >= 0),
  cost_is_estimate      BOOLEAN NOT NULL DEFAULT false,
  latency_ms            INTEGER CHECK (latency_ms >= 0),
  created_at            TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX llm_usage_job ON llm_usage (job_id);  -- FKs are not indexed automatically
CREATE INDEX llm_usage_created ON llm_usage (created_at);

A column default sets updated_at only on insert; maintain it with a trigger or in every UPDATE. Consider row-level security on tenant_id as defence in depth.

API boundary (FastAPI)

Check-then-insert idempotency is racy: two concurrent requests with the same key both miss, and the second insert hits the unique constraint as a 500 error. Insert with ON CONFLICT DO NOTHING; if no row comes back, read the existing row in a separate statement, because under READ COMMITTED a read inside the same statement may not see the concurrent winner. require_authenticated_principal and the repository are interfaces to implement against your identity provider and database; never deploy permissive stubs.

import hashlib
from typing import Annotated, Literal
from uuid import UUID
from fastapi import Depends, FastAPI, Header, HTTPException, Response
from pydantic import BaseModel, ConfigDict, Field

TaskType = Literal["topic_tags", "post_summary", "reply_draft", "weekly_report"]  # = router
app = FastAPI(title="Pulse LLM Jobs")

class CreateJob(BaseModel):
    model_config = ConfigDict(extra="forbid")
    task_type: TaskType
    source_id: str = Field(min_length=1, max_length=128)
    source_version: str = Field(min_length=1, max_length=128)
    text: str = Field(min_length=1, max_length=20_000)

class JobAccepted(BaseModel):
    job_id: UUID
    status: str

@app.post("/v1/llm-jobs", response_model=JobAccepted, status_code=202)
async def create_job(
    body: CreateJob,
    response: Response,
    idempotency_key: Annotated[str, Header(min_length=8, max_length=128)],
    principal: Annotated[Principal, Depends(require_authenticated_principal)],
    repo: Annotated[JobRepository, Depends(get_repository)],
) -> JobAccepted:
    await authorize_source_access(principal, body.source_id)  # 403 unless owner
    request_hash = hashlib.sha256(body.model_dump_json().encode()).hexdigest()
    # Race-free: INSERT ... ON CONFLICT (tenant_id, idempotency_key) DO NOTHING
    # (+ outbox row, same transaction); on conflict, SELECT the existing row.
    job, created = await repo.create_or_get(
        tenant_id=principal.tenant_id, actor_id=principal.actor_id,
        idempotency_key=idempotency_key, request_hash=request_hash, **body.model_dump())
    if not created and job.request_hash != request_hash:
        raise HTTPException(409, "Idempotency-Key reused with a different request body")
    response.headers["Location"] = f"/v1/llm-jobs/{job.job_id}"
    return JobAccepted(job_id=job.job_id, status=job.status)  # actual current status

A replay returns the job's real status and a Location header for polling; reusing a key with a different body returns 409. Also add per-tenant rate limits, gateway body limits, CSRF protection for cookie sessions, and readiness checks on the database and queue.

Worker orchestration

The order is the control. Authorisation and policy run before the cache, so a cache hit can never bypass them. Budget is reserved before the provider call and released on failure. Usage is recorded immediately after the call, before validation can fail, so every billed token is accounted for.

async def handle_job(job, deps) -> None:
    """Orchestration order. Every step is persisted; redelivery must be safe."""
    await deps.authz.verify_source(job.tenant_id, job.source_id)  # access may be revoked
    policy = await deps.policy_engine.inspect(job.input_text, job.policy_version)
    route = choose_route(
        task=job.task_type, risk=policy.risk, uncertainty=policy.uncertainty,
        input_tokens=await deps.tokens.count(job.input_text, job.task_type),
        max_input_tokens=deps.config.context_limits,
        review_threshold=deps.config.review_threshold)
    if route.disposition == "reject":
        return await deps.jobs.mark(job.job_id, "rejected", route.reason)
    if route.disposition == "human_review":
        await deps.jobs.mark(job.job_id, "needs_review", route.reason)
        return await deps.review_queue.enqueue(job.job_id)

    key = await deps.cache.make_key(**deps.cache_scope(job, route.tier))
    cached = await deps.cache.get(key)  # only reached after authz and policy checks
    if cached is not CACHE_MISS:
        return await deps.jobs.complete(job.job_id, cached, source="cache")

    reservation = await deps.budget.reserve(job.tenant_id, route.tier)  # worst case
    try:
        response = await deps.provider.generate(
            tier=route.tier, prompt_version=job.prompt_version,
            text=job.input_text, timeout_s=deps.config.timeout_s)
    except BaseException:
        await deps.budget.release(reservation)
        raise  # queue retries transient errors, then dead-letters
    usage = deps.provider.normalize_usage(response)  # record before anything can fail
    await deps.usage.record(job.job_id, response.model, usage,
                            calculate_cost(route.tier, usage))
    await deps.budget.settle(reservation, usage)

    validated = deps.output_validator.validate(job.task_type, response.text)
    if not validated.accepted:
        return await deps.jobs.mark(job.job_id, "needs_review", "invalid_output")
    await deps.jobs.complete(job.job_id, validated.result, source="model")  # upsert
    await deps.cache.set(key, validated.result)

This is orchestration code over interfaces; transaction boundaries, acknowledgements, lease renewal and cancellation depend on the queue and database you choose. A crash after the provider call but before completion repeats the call on redelivery; accept that cost or persist the raw response first.

8. Budgets, Quotas and Cost Observability

Before dispatch, reserve the worst-case cost (maximum input and output); after the response, settle with actual usage. If the provider omits a usage field, set cost_is_estimate rather than presenting the number as exact.

Limits: per request (input size, output tokens, tool calls, wall-clock time); per user and tenant (daily spend, request rate, concurrency, batch volume); global (provider budget, kill switch, model allowlist, reserved capacity for safety workflows).

Metrics: requests by task and tier, cache hit ratio, uncached, cached and output tokens, cost per success, p50/p95/p99 latency, errors, retries, queue age and human-review rate. Keep Prometheus labels low-cardinality (no user, post or raw tenant IDs); per-tenant cost belongs in the database.

The headline metric, cost per successful workflow, is best computed in SQL:

-- Cost per successful workflow, last 7 days. LEFT JOIN keeps cache-hit jobs
-- (no usage rows) as cheap successes; failed and retried calls still add cost.
SELECT j.task_type,
       count(DISTINCT j.job_id) FILTER (WHERE j.status = 'completed') AS successes,
       coalesce(sum(u.cost_usd), 0)                                  AS cost_usd,
       coalesce(sum(u.cost_usd), 0)
         / NULLIF(count(DISTINCT j.job_id) FILTER (WHERE j.status = 'completed'), 0)
                                                                     AS cost_per_success,
       sum(u.cache_read_tokens)::numeric
         / NULLIF(sum(u.uncached_input_tokens + u.cache_write_tokens
                      + u.cache_read_tokens), 0)                     AS cached_input_share,
       coalesce(bool_or(u.cost_is_estimate), false)                  AS includes_estimates
FROM llm_jobs j
LEFT JOIN llm_usage u USING (job_id)
WHERE j.created_at >= now() - interval '7 days'
GROUP BY j.task_type
ORDER BY cost_usd DESC;

9. Measuring Business Impact

Comparing creators who use the assistant with those who don't is confounded: more active creators adopt more and already respond faster, so that comparison cannot show the assistant caused faster responses. Prefer randomised creator- or account-level experiments with predefined primary and guardrail metrics, duration, minimum detectable effect and stopping rules. Report confidence intervals by language, creator size and surface; monitor reports, appeals, opt-outs, harmful output and human overrides; and never optimise acceptance rate alone, because creators may accept bland or unsafe suggestions.

10. Failure Modes and Operational Response

11. Security, Privacy and Governance

Authenticate every caller and authorise every source object against its tenant; never trust tenant IDs from request bodies. Keep provider keys in a secret manager, rotate them and restrict egress.

Minimise personal data sent to providers and document retention, training-use settings, regional processing and contracts. Encrypt data in transit and at rest, with retention rules for prompts, outputs, traces and backups.

Treat comments, posts, links and attachments as untrusted and defend against prompt injection. Redact logs, and keep audit events with actor, tenant, resource, prompt and policy versions, decision and time.

Keep publishing, account enforcement and other irreversible actions behind explicit permissions and human approval.

12. Testing and Rollout

Every prompt, model, routing, cache and policy change is a deployment. Promote it through offline evaluation, shadow traffic, a 1% canary and staged rollout with automated stop conditions, and keep a tested rollback.

Unit and property tests: cost for each provider's usage shape, cost never negative, tenant A never hits tenant B's cache, route policy, batch state transitions.

Integration and load tests: Redis outage, transaction rollback, queue redelivery, provider 429 and timeouts, concurrent duplicate idempotency keys, partial batch failure, viral bursts, cache stampedes, budget exhaustion.

Security and evaluation tests: broken object-level authorisation, prompt injection, PII leakage; multilingual and adversarial golden sets with moderation false-negative thresholds.

Conclusion

The most durable savings come from optimising the whole workflow, not the price of a token: cache only equivalent work, route by measured quality and risk, batch only what can wait, and choose models with representative evaluations. Pair every optimisation with tenant isolation, budgets, observability, safe fallbacks and human governance. The code here is tested teaching material over declared interfaces, not a turnkey system.

Further reading: current provider pricing and docs on usage fields, prompt caching and batch limits: developers.openai.com/docs/guides/prompt-caching and platform.claude.com/docs/en/build-with-claude/prompt-caching.

Back to all blogs