AskAiml
Sign in Get started

Cost optimisation, streaming, and token hygiene

Model routing, prompt caching, streaming to the UI, and cutting spend without hurting quality.

💬 Module 1 7 min read Not started

Why this matters

LLM cost is now a major line item on every product P&L. Engineers who can cut spend 50%+ without hurting quality become unfireable. The patterns in this lesson — caching, cascading, batching, streaming, compression — collectively give 5-20x cost wins.

Learning objectives

  1. Measure cost per request and per feature.
  2. Apply prompt caching, semantic caching, and batch APIs.
  3. Stream responses to a UI with low latency.
  4. Cascade between cheap and expensive models with confidence routing.
  5. Self-host an open model when API economics break down.

1. Know your cost (instrument first)

You cannot optimise what you do not measure. Wrap every call:

python
def call(prompt: str, model: str = "gpt-4.1-mini") -> dict:
    r = client.responses.create(model=model, input=prompt)
    usage = r.usage
    cost = price(model, usage.input_tokens, usage.output_tokens)
    log({"model": model, "in": usage.input_tokens, "out": usage.output_tokens, "usd": cost})
    return {"text": r.output_text, "usd": cost}

Build a tiny pricing.py:

python
# 2026 published rates (always verify on the model page!)
PRICES = {
    "gpt-4.1":            (2.00, 8.00),     # USD per 1M input/output
    "gpt-4.1-mini":       (0.40, 1.60),
    "gpt-5-nano":         (0.10, 0.40),
    "gpt-5.5":            (5.00, 20.00),
    "claude-opus-4-7":    (15.00, 75.00),
    "claude-sonnet-4-6":  (3.00, 15.00),
    "claude-haiku-4-5":   (0.80, 4.00),
    "gemini-2.5-flash":   (0.10, 0.40),
    "gemini-2.5-pro":     (2.50, 10.00),
}

def price(model: str, in_tokens: int, out_tokens: int) -> float:
    pin, pout = PRICES[model]
    return (in_tokens * pin + out_tokens * pout) / 1_000_000

Send these logs to LangSmith / Langfuse / your own DB. Build a Grafana panel: "$/1000 requests by feature." Done. You can now optimise.

2. The cost optimisation pyramid (top wins first)

2.1 Use a smaller model

You will be surprised how often gpt-4.1-mini matches gpt-5.5 on your task. Run an eval. If accuracy drop is < 2%, ship the small model. 70-90% savings in many cases.

python
def cascaded_call(prompt):
    cheap = call(prompt, model="gpt-4.1-mini")
    if confidence(cheap["text"]) >= 0.85:
        return cheap
    return call(prompt, model="gpt-5.5")

confidence can be a Pydantic-validated confidence field returned by the model itself ("rate your confidence 0-1") plus a heuristic (e.g. tool not called when expected).

2.2 Prompt caching (Anthropic 90% off, OpenAI 50% off)

Already covered in Lesson 1.1. Apply it to every stable system prompt over 1024 tokens.

python
resp = aclient.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=512,
    system=[{"type": "text", "text": LONG_STABLE_SYSTEM, "cache_control": {"type": "ephemeral"}}],
    messages=history,
)

OpenAI caches automatically when prompts share a prefix and the prefix is at least 1024 tokens. You see savings in usage.prompt_tokens_details.cached_tokens.

2.3 Semantic caching

If two users ask "How do I reset my password?" 60 seconds apart, you should cache the answer.

python
# uv add gptcache
from gptcache import cache, Config
from gptcache.adapter.openai import openai as cached_openai
from gptcache.embedding import OpenAI as OAEmb

cache.init(embedding_func=OAEmb().to_embeddings, similarity_threshold=0.92)
cache.set_openai_key()

r = cached_openai.ChatCompletion.create(
    model="gpt-4.1-mini",
    messages=[{"role":"user","content":"How do I reset my password?"}],
)

For agents, also explore langchain.cache (SQLiteCache, RedisSemanticCache).

Caveat: only cache deterministic answers. Cache invalidation is the second hardest problem in CS — set TTLs and bust on data updates.

2.4 Batch APIs (50% discount, async)

OpenAI and Anthropic offer batch endpoints: submit a JSONL of requests, get results within 24h, pay 50% less. Perfect for nightly evaluations, RAG re-indexing, large summarisation jobs.

python
# OpenAI batch
batch = client.batches.create(
    input_file_id=file.id,
    endpoint="/v1/responses",
    completion_window="24h",
)

2.5 Output budgets and stop sequences

max_tokens=512 saves money on runaway responses. Add stop sequences ("```", "") so the model halts on cue.

2.6 Compress long context (LLMLingua)

When you must send a 50k-token document, compress it to 5k with negligible quality loss using LLMLingua-2 (Microsoft, open-source). Trains a small classifier to keep informative tokens.

python
# uv add llmlingua
from llmlingua import PromptCompressor
pc = PromptCompressor(model_name="microsoft/llmlingua-2-xlm-roberta-large-meetingbank")
compressed = pc.compress_prompt(long_text, target_token=4000)

2.7 Self-host with vLLM (when volume justifies)

Once you process > 5M tokens/day on the same model family, self-hosting Llama 3.3 70B, Qwen 2.5 72B, or DeepSeek-V3 on vLLM can be 5-20x cheaper per token than a paid API.

You will learn this in Phase 5. For now, know that vLLM exposes an OpenAI-compatible API so your code does not change:

python
client = OpenAI(base_url="http://my-vllm:8000/v1", api_key="anything")

3. Streaming for snappy UX

Streaming is mostly UX, not cost (you still pay full tokens). But for any chat UI it is mandatory — users see the first token in 200ms instead of waiting 5s for the full answer.

OpenAI Responses streaming

python
with client.responses.stream(
    model="gpt-4.1-mini",
    input="Explain RAG to a 12-year-old.",
) as stream:
    for event in stream:
        if event.type == "response.output_text.delta":
            yield event.delta  # for FastAPI SSE

Anthropic streaming

python
with aclient.messages.stream(
    model="claude-haiku-4-5",
    max_tokens=512,
    messages=[{"role":"user","content":"Explain RAG."}],
) as stream:
    for chunk in stream.text_stream:
        yield chunk

FastAPI Server-Sent Events (the production pattern)

python
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.post("/chat")
def chat(body: dict):
    def gen():
        with client.responses.stream(model="gpt-4.1-mini", input=body["q"]) as s:
            for e in s:
                if e.type == "response.output_text.delta":
                    yield f"data: {e.delta}\n\n"
            yield "data: [DONE]\n\n"
    return StreamingResponse(gen(), media_type="text/event-stream")

Front-end consumes with EventSource (vanilla JS) or react-markdown-streaming.

4. Token hygiene checklist

Before you ship a feature, ask:

  • Is the system prompt cached?
  • Are you using the smallest model that passes evals?
  • Is max_tokens set?
  • Are conversation histories trimmed (tail last N turns + summary)?
  • Are documents chunked + retrieved (RAG) instead of stuffed?
  • Are tool results truncated to a safe max?
  • Are you streaming for UI calls?
  • Are batch jobs using batch APIs?
  • Is each cost logged with feature/route tags?
  • Is there an alert when cost-per-1k-requests rises 20%?

That checklist alone, applied rigorously, beats most "AI cost optimisation" consulting decks.

Hands-on lab (3 hours)

Take your travel agent from Lesson 1.4. Add:

  1. Pricing table + per-call logging.
  2. Prompt caching for the system prompt (Anthropic) and a stable tool catalog block.
  3. Cascaded model: try claude-haiku-4-5 first, escalate to claude-sonnet-4-6 only if confidence < 0.7.
  4. Semantic cache with GPTCache for FAQ-type queries.
  5. Streaming UI in Streamlit so the answer types out word by word.
  6. Generate a cost report that prints: average tokens, average $/query, cache hit rate, model split.

Acceptance criteria:

  • Cache hit rate ≥ 30% on a 50-query test set with intentional repeats.
  • Cascading saves ≥ 40% vs always-Sonnet baseline at <5% accuracy drop.
  • Latency-to-first-token ≤ 500ms.

Common pitfalls

  1. Caching dynamic content. Date in the system prompt = 0% hit rate.
  2. Trusting the model's self-rated confidence blindly. Use it as a hint, combine with heuristics.
  3. Streaming partial JSON to clients. UIs choke. Stream prose; deliver structured data at the end.
  4. Compressing then violating cache. LLMLingua output is non-deterministic — cache the compressed string.
  5. Not paginating tool results. A list_users returning 10k rows kills cost and context.

Self-check

  1. What is the difference between automatic and manual prompt caching in Anthropic?
  2. When does semantic caching go wrong?
  3. How does model cascading interact with eval coverage?
  4. Why does streaming not save cost?
  5. At what monthly token volume does self-hosting Llama 3.3 70B usually beat the API price?

References

Sign in to save your progress and earn badges.