Audit & optimization

Breaking a support call into token line items

A business software vendor runs a support assistant in production. The bill keeps climbing, nobody knows what one answer costs, and next year's budget is decided soon.

Invented caseThis case is not an engagement. It is invented end to end, and the code below is the real code that would run it.

1. Human·The bill keeps climbing

   The assistant has answered well since it went live. The monthly bill keeps climbing and nobody knows why. What does one answer cost?

2. The agent·Inspect a real turn

   An aggregated bill does not say where the volume comes from. So I inspect a real turn, the seventh of a live conversation, and separate what changes at every turn from what does not. That second block is the replayed context.

3. Tool call·The inspection call

   {
     "tool": "inspect_call",
     "conversation": "conv-8412",
     "turn": 7,
     "break_down": true,
     "line_items": [
       "system_prompt",
       "product_catalog",
       "response_examples",
       "conversation_history",
       "customer_question"
     ],
     "compare_with_turn": 6
   }

   The inspection runs on a production call, not one built for the demo.

4. Tool response·The reading for that turn

   {
     "conversation": "conv-8412",
     "turn": 7,
     "tokens_sent": 17740,
     "line_items": {
       "system_prompt": 1180,
       "product_catalog": 9400,
       "response_examples": 2600,
       "conversation_history": 4350,
       "customer_question": 210
     },
     "fingerprints_identical_to_turn_6": [
       "system_prompt",
       "product_catalog",
       "response_examples"
     ],
     "replayed_context": 13180,
     "cache_read_tokens": 0,
     "cache_control_present": false,
     "calls_observed": 1,
     "calls_required": 200,
     "bill_estimate": "refused, sample of a single call"
   }

   The reading refuses to price a bill from a single call: the guard in the code requires the series.

5. Tool call·The replayed request

   {
     "tool": "replay_call",
     "conversation": "conv-8412",
     "turn": 7,
     "blocks": [
       {
         "line_item": "system_prompt",
         "cache_control": "ephemeral"
       },
       {
         "line_item": "response_examples",
         "cache_control": "ephemeral"
       },
       {
         "line_item": "product_catalog",
         "mode": "retrieved_excerpts",
         "records": 3
       },
       {
         "line_item": "conversation_history",
         "turns_kept": 3
       },
       {
         "line_item": "customer_question"
       }
     ],
     "measure": [
       "tokens_sent",
       "cache_read_tokens",
       "similarity_to_original_answer"
     ]
   }

   The cached prefix must stay identical byte for byte, or the cache read never fires.

6. Answer·Volume by line item

   The inspected turn sends 17,740 tokens: 1,180 of system prompt, 9,400 of catalog, 2,600 of examples, 4,350 of history, 210 of question. Three line items carry the same fingerprint as the previous turn and weigh 13,180 tokens, resent in full, with no cache. The replayed request marks those blocks for the cache, cuts the catalog to three records and the history to three turns. Its measurement comes with the series, and no figure lands before that: the token price stays the one in your contract.

"""Token line item reading for a support assistant running in production."""

from collections import Counter

MINIMUM_CALLS = 200
STABLE_LINE_ITEMS = ("system_prompt", "product_catalog", "response_examples")


class SampleTooSmall(Exception):
    """A reading drawn from a handful of calls is contradicted by the first spike."""

def inspect_call(call):
    line_items, fingerprints = Counter(), {}
    for block in call["blocks"]:
        line_items[block["line_item"]] += block["tokens"]
        fingerprints[block["line_item"]] = block["fingerprint"]
    return line_items, fingerprints

def replay_call(call, plan):
    kept = {block["line_item"]: block for block in plan}
    blocks = [dict(b, **kept[b["line_item"]]) for b in call["blocks"] if b["line_item"] in kept]
    return dict(call, blocks=blocks)

def replayed_context(line_items, fingerprints, previous):
    stable = [item for item in line_items if item in STABLE_LINE_ITEMS]
    return sum(line_items[i] for i in stable if previous.get(i) == fingerprints[i])

def report(calls, minimum=MINIMUM_CALLS):
    if len(calls) < minimum:
        raise SampleTooSmall(
            "sample too small: {} of the {} calls required".format(len(calls), minimum)
        )
    total, replayed, cache_read = Counter(), 0, 0
    conversation, previous = None, {}
    for call in sorted(calls, key=lambda c: (c["conversation"], c["turn"])):
        # Replayed context is counted inside one conversation, never across two.
        if call["conversation"] != conversation:
            conversation, previous = call["conversation"], {}
        line_items, fingerprints = inspect_call(call)
        total.update(line_items)
        replayed += replayed_context(line_items, fingerprints, previous)
        cache_read += call.get("cache_read_tokens", 0)
        previous = fingerprints
    sent = sum(total.values())
    return {
        "calls": len(calls),
        "tokens_sent": sent,
        "tokens_per_answer": sent // len(calls),
        "replayed_context": replayed,
        "cache_read_tokens": cache_read,
        "line_items": dict(total.most_common()),
    }

def bill(reading, input_price, cache_read_price):
    # Prices come from the contract in force, never from the code.
    full = reading["tokens_sent"] - reading["cache_read_tokens"]
    return full * input_price + reading["cache_read_tokens"] * cache_read_price

Until someone breaks down a real call, an AI budget is not steered; it is endured.