leanwire

Cut LLM token cost by reshaping the wire format, not the content. Two independent levers, both of which leave what the model actually decides alone.

Zero runtime dependencies. Works with the Anthropic SDK or raw HTTP.

pip install leanwire

1. leanwire.cache — stop re-paying for your transcript

A long agent conversation is re-sent on every turn. If cache_control is only on your system prompt and tools, the static prefix caches and the transcript is billed at full input price on every single call — while your aggregate cache-read numbers look great.

from leanwire.cache import CachePolicy

policy = CachePolicy(ttl="5m", model="claude-opus-4-8")

request = {"model": "claude-opus-4-8", "system": [...], "tools": [...],
           "messages": messages}
placement = policy.apply(request)      # your request is not mutated
response = client.messages.create(**placement.request)

apply() spends whatever breakpoint budget is left after your own tools/system markers (the API allows 4), spacing markers lookback blocks apart from the tail backwards so a valid read point always exists inside the 20-block lookback window — statelessly, with no need to remember where the last request put them.

It refuses to act rather than act wrongly: no budget left, no dict content blocks, or a prefix below the model's minimum cacheable size all produce a no-op with placement.skipped_reason set.

Find out if you have this problem in one loop

from leanwire.cache import CacheAudit

audit = CacheAudit()
for response in your_agent_run():
    audit.observe(response.usage)

print(audit.report)

40 calls | uncached 1,200,000 | read 2,000,000 | write 0 | out 40,000 | 62.5% of input served from cache
  [!] uncached input grows 10,000 -> 50,000 tokens across the run: the conversation
      transcript is being re-billed at full price every call while cache reads stay
      flat -- a static prefix is cached but the messages are not. Place a
      message-level breakpoint

Also detects nothing-cached, write-but-never-read (a timestamp or uuid in your prefix), and bulk prefix rebuilds (TTL expiry). Small per-turn writes are correct and are not flagged.

2. leanwire.codec — stop re-emitting field names

When a model returns N records sharing a schema, it re-emits every key N times.

from leanwire.codec import RecordCodec

codec = RecordCodec.infer(sample_records)     # or build Fields explicitly
codec.verify(sample_records)                  # raises unless round-trip is exact

schema = codec.json_schema()                  # put on output_config.format
prompt_hint = codec.legend()                  # field order + enum codes

records = codec.decode(response_rows)         # back to your original dicts

{"column_name": "loc_na", "score": 10, "criterion_met": true, "hallucination_risk": "low", ...} becomes ["loc_na", 10, true, "l", ...].

Lossless by construction and tested as such: fields that never vary leave the wire and are re-injected on decode, low-cardinality strings become single-character codes, and original key order is restored.

stats = codec.measure(records, token_counter)
print(stats)   # 40 records: 4,860 -> 2,489 tokens (48.8% smaller)

Measure before you promise. Savings depend entirely on how much of your payload is packaging versus free text. In our own testing the same codec gave 49% on records with short scalar fields and 25% on records dominated by long prose -- a 2x spread on identical code. measure() exists so you get a real number on your data rather than an estimate. Never quote a figure you have not run.

3. leanwire.accounting

from leanwire.accounting import cost_of
cost_of(response.usage, "claude-opus-4-8")     # -> Cost(input=..., cache_read=..., ...)

Current first-party prices, with the 1.25x (5m) / 2x (1h) cache-write and 0.1x cache-read multipliers applied.

Which lever applies to you

symptom	lever
Long multi-turn agent, input tokens climbing per call	cache
Cache reads look high but the bill still grows	cache — run CacheAudit
Model returns many records with the same schema	codec
Output is most of your spend	codec
Single short calls, no repetition	neither; measure before optimising

Caveats worth reading

• Cache placement changes billing metadata only — the model sees a byte-identical prompt. It needs no accuracy evaluation.
• The codec changes the output contract. It is lossless in encoding, but you are asking the model to emit a different shape, so evaluate that it still fills the fields correctly on your own data before rolling out.
• Minimum cacheable prefix is model-dependent and not monotonic across generations (512 on Opus 5, 1024 on Opus 4.8, 4096 on Opus 4.6). Pass model= and a token_counter and the policy will skip rather than pay a write that never caches.

License

MIT

Changelog

0.1.1

• Fix (important): RecordCodec.infer() could mis-detect a free-text field as an enum when inferred from a small sample whose values happened to be short. The bogus code list then went into json_schema() and legend(), instructing the model to emit one-letter codes for prose, and decode() mapped those codes back to whichever sample sentence they came from — silently wrong content. verify() did not catch it, because round-tripping the inferred sample really is lossless; it checks losslessness, not whether the schema fits your data. Enum detection now also requires no sentence punctuation, at most 3 words per value, and observed repetition (enum_min_repeat, default 2 records per distinct value). It errs toward "not an enum": worst case you compress a little less. New knobs: enum_max_words, enum_min_repeat.
• json_schema() now accepts array_name positionally as well as by keyword.

0.1.0

• Initial release.

leanwire.lazyprompt — stop re-reading a huge system prompt

A system prompt is resident on every turn. With caching working that is 0.1× per turn, but still per turn — on a 135k-token prompt across 225-turn sessions we measured it at over half of total spend.

from leanwire import LazyPrompt

lp = LazyPrompt.from_markdown(SYSTEM_PROMPT, core=["Non-negotiable rules"])
if not lp.worthwhile(token_counter):        # index has a fixed cost
    ...                                      # small prompts: don't bother

system = lp.resident()                       # core + an index of what exists
tools  = [*your_tools, lp.tool_definition()]
# on a tool call:
result = lp.resolve(block.input["section"])

Pulled sections are not free: they cost a round trip and land in the transcript. estimate(token_counter, turns=..., pulls=[...]) charges that against the all-resident baseline so you can check the trade on your own traces first.

leanwire.compaction — tune how hard you compact

Dropping a transcript token costs 1.25× once and saves 0.1× per remaining turn, so it repays after ~13 turns. Most harnesses compact late and keep too much.

from leanwire import CompactionPolicy, growth_from_usage

policy = CompactionPolicy(threshold_tokens=180_000, keep=0.5,
                          static_tokens=len_of_your_system_block)
if policy.should_compact(prefix_tokens, turns_remaining=n):
    ...

# price a setting on your own usage BEFORE changing anything
growth = growth_from_usage(session_usage, static_tokens=135_000)
print(CompactionPolicy(180_000, 0.5, static_tokens=135_000).simulate(growth))

The policy refuses a threshold below static_tokens — with a 135k system block a 120k threshold is unreachable, and silently clamping it produces numbers for policies you cannot run.

This lever changes what the model can see. Unlike cache placement it is not free; simulate, then A/B against your own quality gate before shipping.

leanwire.parallel — stop paying a prefix read per tool call

The whole prefix is re-read every turn, so cost is roughly turns x prefix. An agent that makes exactly one tool call per turn pays a full prefix read per tool call. Two independent calls in one turn cost one read instead of two.

Multi-call turns are on by default in the API and switched off by tool_choice.disable_parallel_tool_use — easy to set once and never revisit. We measured a production agent running at 0.99 tool calls per turn with that flag on.

from leanwire import ParallelPolicy, TurnStats

request = ParallelPolicy(directive=True).apply(request).request
stats = TurnStats()
stats.observe(response)      # did it actually batch?
print(stats)                 # 412 turns, 731 tool calls (1.77/turn) | batched 58%

The policy removes the flag and adds a short directive on when batching is safe. It will not batch anything itself — only the model knows whether two commands are independent, and a batched write-then-read gets a stale result. This changes behaviour. Ship it behind your quality gate and read TurnStats rather than assuming the permission was taken.

leanwire.attribution — measure what the change earned, not what the bill did

A run gets cheaper for several reasons at once. Diffing two runs' totals and claiming the difference is wrong whenever the runs did different amounts of work, which for an agent is nearly always.

from leanwire import Attribution

report = Attribution(baseline_per_call=0.128, baseline_calls=3000).measure(usages)
print(report)                          # shape of the output, illustrative figures
#   RATE saving (the SDK)    $+31.10   (+13.3% per call)
#   VOLUME saving (not ours) $+166.86  (3,000 -> 1,800 calls)
#   resident-block floor     $55.00

rate_saving is what request shaping earned. volume_saving is real money that belongs to whoever changed the agent loop. Reporting the sum as one number is the mistake this exists to prevent — on a run shaped like the one above it would overstate the library's contribution by 6x.

floor is the resident prefix re-read on every call. No transcript policy can go below it, so it tells you when to stop tuning context and start reducing calls.

leanwire.stack — all of it, in an order that composes

The levers interact. The lazy prompt rewrites system, the batching directive appends to it, and cache breakpoints must land after both — a breakpoint placed before a block that is then appended marks a prefix that no longer exists.

from leanwire import Stack

st = Stack(model="claude-opus-4-8", system=SYSTEM_PROMPT,
           baseline_per_call=your_current_cost_per_call, parallel=True)

request  = st.apply(request)                 # every enabled lever, right order
response = client.messages.create(**request)
st.observe(response)                         # billing, cache health, batching

for block in response.content:               # answers its own section tool
    if (r := st.tool_result(block)):
        tool_results.append(r)

print(st.report())

Only cache placement is on by default, because only cache placement is free. parallel, lazy and compact change what the model sees or does; turn them on one at a time. report().attribution.levers states which levers the requests actually carried, not which you asked for.

Compaction inside Stack uses the prefix size the API reported on your last call rather than an estimate, and cuts only where a transcript may legally be cut — never between a tool_use and its tool_result. If it cannot find a safe boundary it declines and says so, rather than shipping an orphaned result.

0.3.0

• New leanwire.stack, leanwire.parallel, leanwire.attribution.
• The workload module in your build gains a turnkey stack() entry point, preconfigured from your own measured ledger.

0.2.0

• New leanwire.lazyprompt and leanwire.compaction.