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

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.

0.2.0

• New leanwire.lazyprompt and leanwire.compaction.