Token Consumption and Cost Awareness

Module 5, Unit 3 | Lesson 2 of 5

By the end of this lesson, you will be able to:

• Explain what tokens are and how LLM APIs measure them (K6, K11)
• Read token usage from API responses and attach it to your pipeline's stage results (S10, B6)
• Calculate the real cost of running your classifier at scale (K11)
• Apply at least three strategies to reduce token consumption without sacrificing output quality (S10, K6)

In L3.1 you built a classifier that calls an LLM API for every flagged record. That works. But every one of those calls consumes tokens — and tokens cost money. In a production pipeline processing thousands of records per day, unmanaged token consumption is a direct operational cost with no ceiling.

This lesson teaches you to measure what you are spending, understand why, and make deliberate engineering decisions to bring it under control.

What is a token?

LLMs do not process text character by character or word by word. They process tokens — chunks of text that a tokeniser splits your input into before passing it to the model.

🔑 Key term — Token: The basic unit of text that an LLM processes. A token is roughly 3–4 characters of English text, but varies by language and content. The word classification is one token. The phrase tool_name splits into two. A typical English sentence of 15 words is around 20 tokens.

Every API call has two token counts:

• Prompt tokens (also called input tokens): everything you send — system prompt, user message, any context or examples you include
• Completion tokens (also called output tokens): the text the model generates in response

Both are billed. Completion tokens are typically 3–5× more expensive than prompt tokens, because generating text is computationally heavier than reading it.

Reading token usage from the API

Every OpenAI chat completion response includes a usage field. You already receive it — you just may not be logging it.

response = client.chat.completions.create(
    model="gpt-4.1-mini",
    messages=[...],
)

# The usage object is always present
usage = response.usage
print(usage.prompt_tokens)      # e.g. 312
print(usage.completion_tokens)  # e.g. 47
print(usage.total_tokens)       # e.g. 359

The same pattern applies to Anthropic's API:

response = client.messages.create(
    model="claude-haiku-4-5",
    messages=[...],
)

usage = response.usage
print(usage.input_tokens)   # prompt tokens
print(usage.output_tokens)  # completion tokens

Attaching token usage to your pipeline

Your classifier already receives the usage object on every call — it is thrown away. Capture it in classify_record in classifier.py. Only the last line changes: return validate_result(json.loads(raw)) becomes four lines that keep the counts alongside the classification.

def classify_record(record_data: dict[str, Any]) -> dict[str, Any]:
    """
    Send a single record to the LLM for classification.
    Returns a dict with classification, confidence, reasoning and token counts.
    Raises ValueError if the response cannot be parsed or is invalid.
    """
    user_message = f"""Classify this AI usage log record:

log_id: {record_data.get('log_id')}
tool_name: {record_data.get('tool_name')}
task_type: {record_data.get('task_type')}
flagged_issue: {record_data.get('flagged_issue')}
cost_usd: {record_data.get('cost_usd')}
time_saved_mins: {record_data.get('time_saved_mins')}
quality_score: {record_data.get('quality_score')}
human_reviewed: {record_data.get('human_reviewed')}"""

    response = get_client().chat.completions.create(
        model="gpt-4.1-mini",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_message},
        ],
        temperature=0.1,
        max_completion_tokens=200,
        response_format={"type": "json_schema", "json_schema": CLASSIFICATION_SCHEMA},
    )
    choice = response.choices[0]

    if choice.message.refusal:
        raise ValueError(f"Model refused to classify: {choice.message.refusal}")
    if choice.finish_reason == "length":
        raise ValueError("Response truncated before the JSON was complete")

    raw = choice.message.content
    if not raw:
        raise ValueError("Model returned an empty response")

    result = validate_result(json.loads(raw))
    usage = response.usage
    result["_api_tokens_in"] = usage.prompt_tokens
    result["_api_tokens_out"] = usage.completion_tokens
    return result

Underscore-prefixed keys, by convention, mean "pipeline bookkeeping, not model output" — they sit alongside classification, confidence and reasoning without pretending to be part of the classification. classify_with_fallback passes the whole dict through unchanged, so they arrive in stage_classify for free.

Now carry them into the classified record. Replace classify_item in stages.py — it is the L3.1 function with the last two keys added:

def classify_item(item: dict[str, Any]) -> dict[str, Any]:
    """Classify one record. Clean records skip the LLM entirely."""
    record = item["record"]

    if record.flagged_issue == FlaggedIssue.NONE:
        return {
            **item,
            "classification": "none",
            "confidence": 1.0,
            "reasoning": "No issue flagged - classification not required.",
            "route_override": None,
            "model_used": "none",
            "llm_called": False,
        }

    result = with_retry(
        lambda: classify_with_fallback(record_payload(record)),
        max_attempts=3,
        base_delay=1.0,
    )

    return {
        **item,
        "classification": result["classification"],
        "confidence": result["confidence"],
        "reasoning": result["reasoning"],
        "route_override": result.get("route_override"),
        "route_reason": result.get("route_reason"),
        "llm_called": True,
        "model_used": "openai/gpt-4.1-mini",
        "_api_tokens_in": result.get("_api_tokens_in", 0),
        "_api_tokens_out": result.get("_api_tokens_out", 0),
    }

Finally, total them up after run_in_parallel returns, alongside the llm_calls count you already compute. Replace stage_classify in stages.py with this version — it is the L3.1 function with two sums and three metadata keys added:

def stage_classify(state: PipelineState) -> PipelineState:
    """Stage 3: Classify flagged records using the OpenAI classifier."""
    started = time.time()

    items = [{"record": r} for r in state.valid_records]
    results, errs = run_in_parallel(classify_item, items, max_workers=3)
    state.classified_records = results
    llm_calls = sum(1 for r in results if r.get("llm_called"))

    # Sum from the results list — not from a counter mutated inside the workers
    prompt_tokens = sum(r.get("_api_tokens_in", 0) for r in results)
    completion_tokens = sum(r.get("_api_tokens_out", 0) for r in results)

    state.stage_results.append(StageResult(
        stage_name="classify",
        records_in=len(state.valid_records),
        records_out=len(results),
        records_failed=len(errs),
        duration_seconds=round(time.time() - started, 3),
        # run_in_parallel keeps the whole failed item, record object and all.
        # The output payload is JSON, so reduce each failure to the two facts a
        # recipient needs: which record, and what went wrong.
        errors=[
            {"log_id": e["item"]["record"].log_id, "error": e["error"]}
            for e in errs
        ],
        metadata={
            "llm_calls": llm_calls,
            "skipped_clean": len(results) - llm_calls,
            "model": "gpt-4.1-mini",
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens,
            "total_tokens": prompt_tokens + completion_tokens,
        },
    ))

    print(
        f"[classify] {llm_calls} LLM calls | "
        f"{prompt_tokens + completion_tokens} tokens total"
    )
    return state

Curious Cat

Why total the tokens after the parallel run instead of adding them up inside classify_item? Because classify_item executes on several worker threads at once, and total += n is not an atomic operation — two threads can read the same starting value and one increment is lost. The bug is rare, silent, and only appears under concurrency, which makes it miserable to find. Summing over the returned results list afterwards sidesteps the problem entirely: no shared mutable counter, nothing to get wrong. Prefer this shape whenever you can.

Now every pipeline run records exactly how many tokens the classify stage consumed.

Calculating real cost

Knowing your token count is only useful if you translate it into cost. Model names and prices both move — providers reprice regularly and retire models on published shutdown dates, so check the pricing and deprecation pages before you quote a figure or pin a model. The calculation itself never changes:

# Example rates (verify against current pricing pages)
PRICING = {
    "gpt-4.1-mini": {
        "prompt_per_million": 0.40,      # USD per 1M prompt tokens
        "completion_per_million": 1.60,  # USD per 1M completion tokens
    },
    "gpt-4.1": {
        "prompt_per_million": 2.00,
        "completion_per_million": 8.00,
    },
}

def estimate_cost(prompt_tokens: int, completion_tokens: int, model: str) -> float:
    """Return estimated cost in USD."""
    rates = PRICING[model]
    prompt_cost = (prompt_tokens / 1_000_000) * rates["prompt_per_million"]
    completion_cost = (completion_tokens / 1_000_000) * rates["completion_per_million"]
    return round(prompt_cost + completion_cost, 6)

Put estimate_cost in classifier.py next to the pricing table, then use it in stage_classify where you already have the totals:

    # stages.py — after the prompt_tokens / completion_tokens sums
    from classifier import estimate_cost

    estimated_cost = estimate_cost(prompt_tokens, completion_tokens, "gpt-4.1-mini")

Add "estimated_cost_usd": estimated_cost to the same StageResult.metadata dict, and it lands in pipeline_output.json with every run. After a test run on your 30-record dataset you can project daily and monthly costs at your expected record volume — the six cloud calls here will cost a fraction of a penny, which is exactly why you have to multiply before you trust the number.

Coach Cora

Run the cost estimate before you deploy, and do the multiplication yourself. A 400-token prompt with a 50-token answer costs about $0.00024 at gpt-4.1-mini rates, so the six cloud calls in this run cost a fraction of a penny. The same pipeline on 1,500 flagged records a day is about $0.36 a day — still nothing. At 150,000 a day it is $36, and the week someone points it at a full year of back history it becomes a budget line you have to defend. Token costs are easy to underestimate precisely because each individual call is cheap. The risk is volume, not unit price.

Five strategies to reduce token consumption

1. Skip unflagged records

You already do this — records with flagged_issue == "none" skip the LLM entirely. Always filter before you call.

Impact: If 70% of your records are unflagged, you cut LLM calls by 70% with zero quality loss.

2. Compress your prompt

Long system prompts with examples, formatting instructions, and explanations consume prompt tokens on every call. Measure your base prompt token count, then trim it. Remove any text the model does not actually need to classify correctly.

# Before: 420 prompt tokens — a verbose preamble the model does not need
SYSTEM_PROMPT = """
You are a compliance classification assistant for an AI usage monitoring platform.
Your task is to analyse tool usage records that have been flagged...
[100 words of context]
"""

# After: 180 prompt tokens — same output contract, less scaffolding
SYSTEM_PROMPT = """Classify this AI tool usage record into exactly one category:
compliance_risk | operational_issue | cost_anomaly | data_quality

Respond with JSON only:
{"classification": "<category>", "confidence": <0.0-1.0>, "reasoning": "<one sentence>"}"""

Keep all three output keys. classification, confidence and reasoning are what the validation guards in classify_record check for, and reasoning is what makes a classification auditable in L3.4's logs — trimming it would save a handful of completion tokens and cost you your audit trail. Compress the instructions, never the contract.

It is tempting to go further and delete the Respond with JSON only: {...} block too, on the grounds that the strict schema you added in L3.1 already enforces it. Resist that one, and it is worth knowing why. In L3.3 this system prompt moves into a shared module and the same text is sent to a local model through Ollama, where format: "json" constrains the syntax but nothing tells the model which keys you want — Ollama's own documentation says you must still instruct the model to respond in JSON. Delete those lines and the cloud path keeps working while the local path starts returning well-formed objects with keys of its own invention. That is the subtler form of the same rule: the contract is not just what this provider enforces, it is what every consumer of the prompt relies on.

Impact: Halving your system prompt from 400 to 200 tokens saves 200 tokens per call. At 10,000 calls per day, that is 2 million prompt tokens saved — roughly $0.80/day at gpt-4.1-mini rates. Small per call; significant at scale.

3. Choose the right model for the task

Not every call needs your most capable model. A structured classification task with a clear schema is well within the capability of smaller, cheaper models.

Both columns are here because a prompt-only comparison hides the thing that often decides the bill. Work your own case out rather than assuming: at the 400-token prompt you start with and a 50-token answer, gpt-4.1-mini costs $0.00016 in and $0.00008 out, so the prompt is twice the completion. Compress that prompt to 180 tokens, as this lesson is about to ask you to, and the completion becomes the larger half. The optimisation moves which column you should be watching, which is the sort of thing you only notice if you priced both.

Read the "best for" column as a tier within a provider rather than across the table. A cheap model from one provider and a cheap model from another are not priced on the same scale, so comparing rows across providers tells you very little about capability.

Route easy cases to a cheaper model and only escalate difficult ones (low-confidence results) to a more powerful model.

4. Batch with confidence routing

Classify with a cheap model first. If confidence is high (≥ 0.9), accept the result. If confidence is low (< 0.9), re-classify with a more capable model.

To do this you need classify_record to accept the model as an argument. Give it a default so nothing else has to change:

# classifier.py — add a model parameter
def classify_record(record_data: dict[str, Any], model: str = "gpt-4.1-mini") -> dict[str, Any]:
    ...
    response = get_client().chat.completions.create(
        model=model,          # was hard-coded to "gpt-4.1-mini"
        ...
    )
    ...
    return result


ESCALATION_THRESHOLD = 0.9


def classify_with_escalation(record_data: dict[str, Any]) -> dict[str, Any]:
    """Cheap model first; escalate to a stronger one only when it is unsure."""
    result = classify_record(record_data, model="gpt-4.1-mini")

    if result["confidence"] >= ESCALATION_THRESHOLD:
        return result

    escalated = classify_record(record_data, model="gpt-4.1")
    escalated["_escalated_from"] = "gpt-4.1-mini"
    return escalated

Note how this stacks with the CONFIDENCE_THRESHOLD of 0.75 from L3.1 rather than replacing it. You now have two thresholds doing two different jobs: below 0.9, try a better model; if the better model is still below 0.75, give up on automation and send the record to a human. Escalation is a cost decision, the confidence threshold is a governance decision, and they are deliberately not the same number.

Impact: If 80% of records are classified confidently by the cheap model, you pay cheap-model prices for 80% of your volume.

5. Cache repeated prompts

If the same record passes through your pipeline on multiple runs — after a failure, or during the benchmarking you will do in L4.2 — caching the classification avoids a redundant API call entirely. The cheapest token is the one you never send.

You will build this properly in L4.2, as a thread-safe ClassifierCache with atomic writes, because a cache shared across parallel workers has failure modes an in-memory dict does not. For now the point is the arithmetic: a cache hit costs zero tokens, so on a re-run of this dataset your classify stage spend drops to nothing.

Two things worth knowing before you get there. A cache is only safe if its key covers everything that could change the answer — key on the record alone and you will serve stale classifications after you improve the prompt. And separately, OpenAI applies server-side prompt caching automatically to long repeated prefixes, which discounts the prefix rather than skipping the call. Note the eligibility rule, because it decides whether the feature applies to you at all: caching only engages at 1,024 prompt tokens or more. Above that floor, cached input is discounted rather than free and the size of the discount varies by model — for gpt-4.1-mini it is $0.10 against $0.40, a 75% saving — and an entry lives around 30 minutes, with the clock reset each time the prefix is reused. A 400-token system prompt like the one in this lesson sits well under the floor and will never cache, however often you send it. Verify rather than assume — the usage object reports how many tokens were actually served from cache.

Adding a cost guardrail

A guardrail stops the pipeline if token spend exceeds a configured threshold per run — protecting against runaway costs from unexpected data volumes.

Use the same halt mechanism as the 10% validation threshold from L1.3: set state.halted and state.halt_reason rather than raising. A raised exception kills the process and loses the StageResult audit record; a halt stops the pipeline and leaves an explanation behind. Add this at the end of stage_classify, once you have the token totals:

MAX_TOKENS_PER_RUN = 500_000  # configurable


    # stages.py — in stage_classify, after computing prompt_tokens/completion_tokens
    #             and appending the StageResult
    total_tokens = prompt_tokens + completion_tokens
    if total_tokens > MAX_TOKENS_PER_RUN:
        state.halted = True
        state.halt_reason = (
            f"Token guardrail exceeded: {total_tokens:,} tokens consumed "
            f"(limit {MAX_TOKENS_PER_RUN:,})"
        )
        print(f"[classify] HALTING: {state.halt_reason}")

Because the runner checks state.halted after every stage, that is all you need — routing, approval and output are skipped, and the run stops with its reason recorded.

This guardrail is a post-hoc check: it fires after the batch has already been paid for. That is the honest limitation of measuring at the stage boundary, and it is fine for a batch of 30. For a batch of 30,000 you would want the check inside the worker, cancelling remaining futures once the budget is gone — which is a genuinely harder problem, because your workers are running concurrently and each holds a call already in flight.

Curious Cat

What should you do when the guardrail fires? You have three options: stop and alert, checkpoint progress and resume later, or route remaining records to a cheaper fallback model. The right choice depends on whether the pipeline is real-time or batch, whether partial results are acceptable, and whether a human needs to approve the overspend. The guardrail itself is non-negotiable in production — the response to it is an engineering decision.

Build activity

1. Capture response.usage in classify_record in classifier.py, returning _api_tokens_in and _api_tokens_out alongside the classification.
2. Pass both through classify_item in stage_classify, then sum them from the results list after run_in_parallel and add prompt_tokens, completion_tokens and total_tokens to StageResult.metadata.
3. Add PRICING and estimate_cost() to classifier.py, and add estimated_cost_usd to the same metadata dict. Run python pipeline.py and read the real figure out of pipeline_output.json.
4. Measure your current system prompt token count using the tiktoken library (pip install tiktoken), then reduce it by at least 20% while keeping all three output keys. Re-run and confirm the classifications are unchanged and prompt_tokens has dropped.
5. Add the MAX_TOKENS_PER_RUN guardrail using state.halted. Set it deliberately low (say 500) and confirm the pipeline halts with the reason recorded; then restore it.
6. Implement classify_with_escalation — cheap model first, escalate below 0.9 confidence. Report total tokens, estimated cost, and how many records escalated.

# Counting tokens in a prompt without making an API call
import tiktoken

# tiktoken maps model names to encodings, but that table lags new releases:
# encoding_for_model raises KeyError for any model it has not been taught yet.
# gpt-4.1-* is known, so this resolves — keep the fallback anyway, because the
# next model you try is exactly the one that will not be.
try:
    enc = tiktoken.encoding_for_model("gpt-4.1-mini")
except KeyError:
    enc = tiktoken.get_encoding("o200k_base")

tokens = enc.encode(your_system_prompt)
print(f"System prompt: {len(tokens)} tokens")

Challenge Chase

Prompt caching is automatic, discounted rather than free, and only applies above 1,024 prompt tokens — so your compressed system prompt does not qualify. Work out what would have to change for it to pay. Read the cached-token field in the usage object and log it alongside the counts you already capture, then answer two questions with arithmetic rather than intuition: how much shared prefix would you need before caching beats the uncached rate, and what would you have to put in that prefix to get there without simply padding it? If the honest answer is that this pipeline will never qualify, that is a legitimate finding — say so in your L4.3 narrative rather than adding a feature that cannot fire.

The code from this lesson

The pipeline with token counting and cost estimation. The escalation pattern and the token guardrail are shown later in the lesson as patches rather than whole functions, so they are not in the archive — the README says so too, and L3.3 carries the guardrail forward.

Take it if you would rather read and run the code than type it out. Typing it is still the better way to learn it — this is here so a typo cannot cost you the lesson.

Checklist

• classify_record returns _api_tokens_in and _api_tokens_out from response.usage
• stage_classify sums them from the results list — not from a counter mutated inside the worker threads
• pipeline_output.json shows prompt_tokens, completion_tokens, total_tokens and estimated_cost_usd in the classify stage metadata
• System prompt has been measured with tiktoken and trimmed by at least 20%, with all three output keys intact
• The MAX_TOKENS_PER_RUN guardrail halts the run via state.halted, not by raising
• classify_with_escalation sends low-confidence records to a more capable model
• I can explain the cost difference between prompt tokens and completion tokens
• I can explain why the escalation threshold (0.9) and the human-review threshold (0.75) are different numbers

Quick Check

KSB evidence focus

• K6 — Understands the tools, techniques and methods used in AI/ML model development and deployment. Token consumption is a core operational metric for any LLM-powered system. Understanding how tokens are counted, where they come from, and how model selection affects cost is foundational knowledge for deploying AI pipelines responsibly.

• K11 — Understands relevant data governance, data privacy and security issues. Cost is a governance concern. Uncontrolled token spend in a production system represents a financial risk that must be monitored, capped, and reported. The guardrail pattern and cost logging you implement here are the technical equivalent of a budget control in a financial system.

• S10 — Can carry out data processing tasks, including processing/cleaning, data transformations, and feature engineering. The routing strategy — cheap model for high-confidence cases, expensive model for escalations — is a data processing decision. It transforms a flat LLM-call-per-record approach into a tiered processing pipeline that balances cost and quality.

Up next: Lesson 3 covers running models locally with Ollama — which eliminates token costs entirely for sensitive records that cannot leave your infrastructure.