Harden estimator/CLI, 4x tests (43->202), 4x demos (5->20)

demos/06_chat_transcript_cost.py

@@ -0,0 +1,45 @@
+"""Scenario 6 - conversational / chatbot products.
+
+A chat product pays for the *whole* transcript on every turn, because each new
+user message re-sends the accumulated history. This demo measures a real
+multi-turn conversation, then shows the compounding cost of carrying full
+history vs. a windowed / summarized context.
+
+Audience: teams shipping chat assistants who need to reason about history cost.
+"""
+from _common import rule, read_fixture, usd
+from tokenmeter.core import estimate
+
+
+def main() -> None:
+    rule("CHAT TRANSCRIPT COST  -  you re-pay for history on every turn")
+
+    convo = read_fixture("05-chat-transcript", "conversation.txt")
+    model = "claude-haiku"
+
+    # Full transcript resent each turn, ~150-token reply.
+    full = estimate(convo, model=model, output_tokens=150)
+    d = full.to_dict()
+    print(f"\nFull transcript on {model} (150-token reply):")
+    print(f"  history_tokens={d['input_tokens']}  per_turn={usd(d['total_cost_usd'])}")
+
+    # A 10-turn session where history grows linearly: turn k pays ~k/N of the
+    # full history. Approximate the running cost.
+    turns = 10
+    running = sum(
+        estimate(convo, model=model, input_tokens=int(full.input_tokens * (k + 1) / turns),
+                 output_tokens=150).total_cost
+        for k in range(turns)
+    )
+    print(f"  {turns}-turn session (history grows each turn): {usd(running)}")
+
+    # Windowed alternative: cap history at a third of the transcript.
+    windowed = estimate(model=model, input_tokens=full.input_tokens // 3, output_tokens=150)
+    saved = full.total_cost - windowed.total_cost
+    print(f"\nCap history to 1/3 the transcript: {usd(windowed.total_cost)}/turn "
+          f"({usd(saved)} saved/turn).")
+    print("Windowing or summarizing history is the biggest lever on chat cost.")
+
+
+if __name__ == "__main__":
+    main()

demos/07_context_window_guard.py

@@ -0,0 +1,35 @@
+"""Scenario 7 - platform guardrails.
+
+Before you send an assembled prompt you want to know it will *fit* the model's
+context window — a request that overflows is rejected by the API after you have
+already paid to build it. This demo takes a document dump and checks it against
+each model's window, showing which models can take it and which cannot.
+
+Audience: platform engineers adding pre-flight window checks.
+"""
+from _common import rule, read_fixture, usd
+from tokenmeter.core import estimate, check_budget, list_models
+
+
+def main() -> None:
+    rule("CONTEXT WINDOW GUARD  -  will this prompt even fit?")
+
+    dump = read_fixture("06-context-window-guard", "document_dump.txt")
+    print("\nA large document dump checked against every model's window:\n")
+    print(f"  {'model':<14}{'window':>10}{'used_%':>10}  fits?")
+    for m in list_models():
+        est = estimate(dump, model=m.name, output_tokens=500)
+        res = check_budget(est)
+        fits = "yes" if res.ok else "NO"
+        d = est.to_dict()
+        print(f"  {m.name:<14}{m.context_window:>10}{d['context_used_pct']:>10}  {fits}")
+
+    smallest = min(list_models(), key=lambda m: m.context_window)
+    est = estimate(dump, model=smallest.name, output_tokens=500)
+    print(f"\nSmallest window ({smallest.name}, {smallest.context_window}) "
+          f"needs {est.input_tokens + est.output_tokens} tokens — "
+          f"pre-flight the check, don't pay to be rejected. cost would be {usd(est.total_cost)}.")
+
+
+if __name__ == "__main__":
+    main()

demos/08_diff_review_cost.py

@@ -0,0 +1,35 @@
+"""Scenario 8 - AI code review in CI.
+
+An automated PR reviewer sends the diff to an LLM on every push. Big diffs cost
+more and can blow the window. This demo prices a real diff and projects the
+monthly bill for a team's PR volume, then shows the cheap-model alternative.
+
+Audience: dev-tooling teams wiring LLM code review into pipelines.
+"""
+from _common import rule, read_fixture, usd
+from tokenmeter.core import estimate, compare_models
+
+
+def main() -> None:
+    rule("DIFF REVIEW COST  -  what does an AI PR reviewer cost per month?")
+
+    diff = read_fixture("09-stdin-pipeline", "sample_diff.txt")
+    print("\nOne PR diff reviewed (~600-token review comment):\n")
+
+    ranking = compare_models(diff, output_tokens=600,
+                             models=["claude-haiku", "claude-sonnet", "gpt-4o", "claude-opus"])
+    for e in ranking:
+        d = e.to_dict()
+        print(f"  {d['model']:<14} per_review={usd(d['total_cost_usd'])}  in={d['input_tokens']}")
+
+    prs_per_day, workdays = 40, 21
+    monthly = prs_per_day * workdays
+    cheap, dear = ranking[0], ranking[-1]
+    print(f"\nAt {prs_per_day} PRs/day x {workdays} workdays = {monthly} reviews/month:")
+    print(f"  {cheap.model}: {usd(cheap.total_cost * monthly)}/mo   "
+          f"{dear.model}: {usd(dear.total_cost * monthly)}/mo")
+    print("Route routine reviews to the cheap model; reserve the expensive one for flagged PRs.")
+
+
+if __name__ == "__main__":
+    main()

demos/09_fewshot_tax.py

@@ -0,0 +1,36 @@
+"""Scenario 9 - prompt engineering trade-offs.
+
+Few-shot examples improve quality but you pay for them on *every* call, forever.
+This demo quantifies the recurring "few-shot tax" against the zero-shot baseline
+and shows the break-even volume where trimming examples pays for itself.
+
+Audience: prompt engineers deciding how many examples to carry.
+"""
+from _common import rule, read_fixture, usd
+from tokenmeter.core import estimate
+
+
+def main() -> None:
+    rule("FEW-SHOT TAX  -  the recurring cost of carrying examples")
+
+    fewshot = read_fixture("07-fewshot-vs-zeroshot", "fewshot.txt")
+    zeroshot = read_fixture("07-fewshot-vs-zeroshot", "zeroshot.txt")
+    model = "claude-sonnet"
+
+    fs = estimate(fewshot, model=model, output_tokens=80)
+    zs = estimate(zeroshot, model=model, output_tokens=80)
+    delta = fs.total_cost - zs.total_cost
+
+    print(f"\nOn {model} (80-token answer):")
+    print(f"  zero-shot: {zs.input_tokens:>4} tok  {usd(zs.total_cost)}")
+    print(f"  few-shot:  {fs.input_tokens:>4} tok  {usd(fs.total_cost)}")
+    print(f"  per-call few-shot premium: {usd(delta)}")
+
+    for volume in (10_000, 100_000, 1_000_000):
+        print(f"  at {volume:>9,} calls: ${delta * volume:,.2f} extra")
+    print("If few-shot doesn't lift quality enough to justify that tax, move examples "
+          "to a fine-tune or a retrieval step.")
+
+
+if __name__ == "__main__":
+    main()

demos/10_classifier_router.py

@@ -0,0 +1,39 @@
+"""Scenario 10 - model routing / cascades.
+
+Many workloads (classification, routing, intent detection) are cheap enough to
+run on a small model, escalating to a big model only on low confidence. This
+demo prices a classifier prompt on a small vs large model and computes the
+blended cost of a cascade that escalates 15% of the time.
+
+Audience: engineers designing model cascades / routers.
+"""
+from _common import rule, read_fixture, usd
+from tokenmeter.core import estimate
+
+
+def main() -> None:
+    rule("CLASSIFIER ROUTER  -  small model first, escalate the hard 15%")
+
+    prompt = read_fixture("03-model-selection", "classifier_prompt.txt")
+    small = estimate(prompt, model="gpt-4o-mini", output_tokens=20)
+    large = estimate(prompt, model="claude-opus", output_tokens=20)
+
+    print(f"\nClassifier prompt (20-token label):")
+    print(f"  small (gpt-4o-mini): {usd(small.total_cost)}")
+    print(f"  large (claude-opus): {usd(large.total_cost)}")
+
+    escalate = 0.15
+    # cascade pays the small model always, and the large model on escalations
+    blended = small.total_cost + escalate * large.total_cost
+    always_large = large.total_cost
+    saved_pct = (1 - blended / always_large) * 100 if always_large else 0
+    print(f"\nCascade (small always + large on {escalate:.0%}): {usd(blended)}/call")
+    print(f"  vs always-large {usd(always_large)}/call -> {saved_pct:.0f}% cheaper")
+
+    calls = 5_000_000
+    print(f"  over {calls:,} calls: ${blended * calls:,.2f} vs ${always_large * calls:,.2f}")
+    print("Cascades win when the small model handles the easy majority correctly.")
+
+
+if __name__ == "__main__":
+    main()

demos/11_agent_step_budget.py

@@ -0,0 +1,38 @@
+"""Scenario 11 - autonomous agents.
+
+An agent loops: think -> act -> observe, many LLM calls per task. Runaway loops
+are a real cost incident. This demo prices one agent step and enforces a
+per-task budget gate that would halt the agent before it overspends.
+
+Audience: teams running autonomous / tool-using agents.
+"""
+from _common import rule, read_fixture, usd
+from tokenmeter.core import estimate, check_budget
+
+
+def main() -> None:
+    rule("AGENT STEP BUDGET  -  cap the loop before it runs away")
+
+    step = read_fixture("08-csv-finops", "agent_step_prompt.txt")
+    model = "claude-sonnet"
+    per_step = estimate(step, model=model, output_tokens=120)
+    print(f"\nOne agent step on {model}: {usd(per_step.total_cost)} "
+          f"({per_step.input_tokens} in / {per_step.output_tokens} out)")
+
+    max_task_cost = 0.05
+    max_steps = int(max_task_cost / per_step.total_cost) if per_step.total_cost else 0
+    print(f"\nPer-task budget {usd(max_task_cost)} => at most ~{max_steps} steps.")
+
+    for n_steps in (10, max_steps, max_steps + 20):
+        projected = estimate(model=model,
+                             input_tokens=per_step.input_tokens * n_steps,
+                             output_tokens=per_step.output_tokens * n_steps)
+        res = check_budget(projected, max_cost_usd=max_task_cost)
+        status = "OK" if res.ok else "HALT"
+        print(f"  {n_steps:>3} steps -> {usd(projected.total_cost)}  [{status}]")
+
+    print("Gate cumulative task cost each step; halt the agent when the ceiling is hit.")
+
+
+if __name__ == "__main__":
+    main()

demos/12_prompt_diet.py

@@ -0,0 +1,38 @@
+"""Scenario 12 - prompt optimization ("prompt diet").
+
+Trimming boilerplate from a system prompt cuts cost on every single call. This
+demo measures a verbose prompt, a trimmed version, and reports the savings both
+per-call and at scale — the ROI of the editing work.
+
+Audience: anyone optimizing a high-volume prompt.
+"""
+from _common import rule, read_fixture, usd
+from tokenmeter.core import estimate, count_tokens
+
+
+def main() -> None:
+    rule("PROMPT DIET  -  measure the ROI of trimming a prompt")
+
+    verbose = read_fixture("01-basic", "prompt.txt")
+    # A trimmed variant: drop the trailing half (simulates removing boilerplate).
+    lines = verbose.splitlines()
+    trimmed = "\n".join(lines[: max(1, len(lines) // 2)])
+
+    model = "claude-sonnet"
+    v = estimate(verbose, model=model, output_tokens=200)
+    t = estimate(trimmed, model=model, output_tokens=200)
+    saved = v.total_cost - t.total_cost
+
+    print(f"\nOn {model} (200-token answer):")
+    print(f"  verbose: {count_tokens(verbose):>4} tok  {usd(v.total_cost)}")
+    print(f"  trimmed: {count_tokens(trimmed):>4} tok  {usd(t.total_cost)}")
+    print(f"  saved per call: {usd(saved)}")
+
+    daily = 200_000
+    print(f"\nAt {daily:,} calls/day the trim saves ${saved * daily:,.2f}/day "
+          f"(${saved * daily * 30:,.2f}/mo).")
+    print("A 10-minute edit can pay for itself many times over at volume.")
+
+
+if __name__ == "__main__":
+    main()

demos/13_rag_chunk_tuning.py

@@ -0,0 +1,33 @@
+"""Scenario 13 - RAG retriever tuning (top-k sweep).
+
+The retriever's top-k directly drives context size and cost. This demo sweeps k
+by measuring what a growing slice of retrieved context costs and where it starts
+eating dangerous fractions of the window.
+
+Audience: RAG engineers picking a top-k / chunk budget.
+"""
+from _common import rule, read_fixture, usd
+from tokenmeter.core import estimate, get_pricing
+
+
+def main() -> None:
+    rule("RAG CHUNK TUNING  -  sweep top-k against cost and window")
+
+    context = read_fixture("02-rag-context", "retrieved_context.txt")
+    model = "claude-sonnet"
+    full_tokens = estimate(context, model=model).input_tokens
+    window = get_pricing(model).context_window
+
+    print(f"\nRetrieved-context budget on {model} (window {window:,}), 300-token answer:\n")
+    print(f"  {'k (fraction)':<14}{'in_tok':>8}{'cost':>14}{'window_%':>12}")
+    for frac in (0.25, 0.5, 0.75, 1.0, 1.5):
+        in_tok = int(full_tokens * frac)
+        est = estimate(model=model, input_tokens=in_tok, output_tokens=300)
+        d = est.to_dict()
+        print(f"  {frac:<14}{in_tok:>8}{usd(d['total_cost_usd']):>14}{d['context_used_pct']:>12}")
+
+    print("\nPick the smallest k that answers well; each extra chunk is paid on every query.")
+
+
+if __name__ == "__main__":
+    main()

demos/14_multi_model_portfolio.py

@@ -0,0 +1,39 @@
+"""Scenario 14 - portfolio cost allocation.
+
+A platform runs several workloads on different models. This demo builds a
+portfolio (each workload = a prompt fixture + a model + a daily volume) and rolls
+up the projected daily and monthly spend, so finance sees one defensible number.
+
+Audience: platform owners doing chargeback / showback.
+"""
+from _common import rule, read_fixture, usd
+from tokenmeter.core import estimate
+
+
+WORKLOADS = [
+    ("support-triage",   ("01-basic", "prompt.txt"),                 "claude-haiku",  200_000, 300),
+    ("rag-answers",      ("02-rag-context", "retrieved_context.txt"), "claude-sonnet",  50_000, 300),
+    ("agent-planning",   ("08-csv-finops", "agent_step_prompt.txt"),  "gpt-4o",        400_000, 120),
+    ("code-review",      ("09-stdin-pipeline", "sample_diff.txt"),    "claude-opus",     5_000, 600),
+]
+
+
+def main() -> None:
+    rule("MULTI-MODEL PORTFOLIO  -  one rollup for finance")
+
+    print(f"\n  {'workload':<18}{'model':<14}{'calls/day':>11}{'$/day':>14}")
+    total_day = 0.0
+    for name, fx, model, volume, out in WORKLOADS:
+        est = estimate(read_fixture(*fx), model=model, output_tokens=out)
+        day = est.total_cost * volume
+        total_day += day
+        print(f"  {name:<18}{model:<14}{volume:>11,}{usd(day):>14}")
+
+    print(f"  {'-' * 55}")
+    print(f"  {'TOTAL':<18}{'':<14}{'':>11}{usd(total_day):>14}")
+    print(f"\nProjected spend: ${total_day:,.2f}/day  ->  ${total_day * 30:,.2f}/mo. "
+          "Re-run when a prompt or volume changes to catch cost drift early.")
+
+
+if __name__ == "__main__":
+    main()

demos/15_streaming_output_forecast.py

@@ -0,0 +1,36 @@
+"""Scenario 15 - output-length sensitivity.
+
+Output tokens are usually the most expensive tokens (output price > input price).
+This demo holds the prompt fixed and sweeps expected output length, showing how
+much a chatty model or an unbounded max_tokens costs you.
+
+Audience: engineers setting max_tokens / controlling verbosity.
+"""
+from _common import rule, read_fixture, usd
+from tokenmeter.core import estimate, get_pricing
+
+
+def main() -> None:
+    rule("OUTPUT LENGTH FORECAST  -  output tokens are the pricey ones")
+
+    prompt = read_fixture("01-basic", "prompt.txt")
+    model = "claude-opus"
+    p = get_pricing(model)
+    print(f"\n{model}: input ${p.input_per_1k}/1k, output ${p.output_per_1k}/1k "
+          f"(output is {p.output_per_1k / p.input_per_1k:.0f}x input)\n")
+
+    print(f"  {'max output tok':<16}{'total cost':>14}")
+    base = None
+    for out in (100, 500, 1000, 2000, 4000):
+        est = estimate(prompt, model=model, output_tokens=out)
+        if base is None:
+            base = est.total_cost
+        print(f"  {out:<16}{usd(est.total_cost):>14}")
+
+    big = estimate(prompt, model=model, output_tokens=4000)
+    print(f"\nGoing from 100 to 4000 output tokens is {big.total_cost / base:.1f}x the cost. "
+          "Cap max_tokens and ask for concise answers where you can.")
+
+
+if __name__ == "__main__":
+    main()