monster/lessismore
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: monster:cda94780705ef66e21523545862370624113bb67
Branches Tags
monster:main
..
compare: monster:493c3e912a3258920a14ba3a30c86fe531ba99da
Branches Tags
monster:main

No commits in common. "cda94780705ef66e21523545862370624113bb67" and "493c3e912a3258920a14ba3a30c86fe531ba99da" have entirely different histories.
cda9478070 ... 493c3e912a

16 changed files with 653 additions and 1050 deletions
Show Stats Expand all files Collapse all files

1
.claude/scheduled_tasks.lock Normal file
View File

@ -0,0 +1 @@
{"sessionId":"b1485757-c727-45be-be51-c0147d9825ad","pid":63486,"procStart":"Tue Jul 7 02:45:21 2026","acquiredAt":1783393961121}

4
.gitignore vendored
View File

@ -1,6 +1,2 @@
__pycache__/
.claude/
build/
*.egg-info/
grunts/
loghub_samples/

315
README.md
View File

@ -1,31 +1,22 @@
# lessismore 📉
**Turn 87,000 tokens of log spam into 6,700 tokens of pure signal — before it
**Turn 50,000 tokens of log spam into 4,500 tokens of pure signal — before it
ever hits your LLM.**
lessismore is a deterministic compression filter for the bulk text that
actually fills context windows: logs, CI output, tracebacks, JSON dumps,
captured tool output. Pure stdlib Python. Zero dependencies. Every claim below
was measured against the real o200k tokenizer, on three public benchmarks as
well as synthetic ones — and every idea that failed the measurement is
documented in the graveyard section, so you know exactly what you're getting.
was measured against the real o200k tokenizer — and every idea that failed the
measurement is documented at the bottom, so you know exactly what you're
getting.
```bash
tail -5000 app.log | python3 lessismore.py | llm "why did this crash?"
tail -5000 app.log | python3 lessismore.py -l 2 | llm "why did this crash?"
```
**13x** on mixed error logs · **16x** on ANSI CI logs · **98%** on captured
pip/tqdm output · **3.9x** on real LogHub system logs · **95% critical-signal
retention** on the LogDx-CI benchmark's 35 real GitHub Actions failures — and
at an equal 2k-token budget, compress-then-truncate keeps **1.9x more**
diagnostic signals than plain truncation · **~6–12 MB/s** single-core ·
**0** dependencies
And the part a compression ratio can't prove: in the root-cause eval
([eval.py](eval.py)), the model found the planted root cause in **8/8**
compressed logs vs **6/8** raw — smaller *and* more often right. On 12 real
LogDx-CI failures it scored 12/12 on both raw and compressed, at less than
half the tokens.
**11x** on mixed error logs · **12x** on interleaved service logs ·
**8.9x** on ANSI CI logs · **98%** on captured pip/tqdm output ·
**~6 MB/s** single-core · **0** dependencies
## Why this exists
@ -46,15 +37,10 @@ Three payoffs when you pipe through it:
- **Context capacity** — an hour of log history fits where two minutes did.
On a local model, that's the difference between full speed and crawling.
- **Prompt-cache longevity** — every pass is deterministic and idempotent:
same input, same output, byte for byte (it's a test invariant). This
matters wherever a prompt is *re-generated* from source each call — a CI
assistant re-reading the same log, RAG context rebuilt per request, a hook
re-running on every tool call. A deterministic pass produces the identical
prefix every time, so the provider cache hits; an ML compressor in the
loop busts it on every subtle variation. (Within a single chat, follow-ups
hit the cache regardless — the transcript is append-only. The claim is
about rebuilt prompts, not chat turns.)
- **Prompt-cache longevity** — every pass is deterministic: same input, same
output, byte for byte. Follow-up questions re-hit the provider cache
instead of re-paying for the logs. An ML compressor in the loop would bust
the cache on every subtle variation; the regex passes never do.
- **Model attention** — LLMs lose things in the middle of walls of text. Feed
signal, not noise, and the first answer is the right one more often.
@ -67,12 +53,10 @@ python3 test_lessismore.py # should print "ok"
```
```bash
lm dump.log > small.log # file in, file out (level 2 is the default)
docker logs myapp 2>&1 | lm # pipe filter
lm --run "pytest -x" # run a noisy command, print it compressed,
# keep its exit code (pipes can't)
lm big.txt --budget 4000 # compress, then hard-cap at ~4k tokens
lm big.txt --ml 0.5 # + LLMLingua last mile
lm dump.log -l 2 > small.log # file in, file out
docker logs myapp 2>&1 | lm -l 2 # pipe filter
lm big.txt -l 2 --budget 4000 # compress, then hard-cap at ~4k tokens
lm big.txt -l 2 --ml 0.5 # + LLMLingua last mile
lm --serve # paste-in demo page on localhost:7777
```
@ -93,37 +77,17 @@ Four levels, from byte-cautious to caveman. Pick by content, not by greed.
| Level | What it eats | Point it at | Profile |
|---|---|---|---|
| **1** | whitespace runs, consecutive duplicate lines | code, scripts, anything | structure-safe: indentation and content untouched (whitespace *inside string literals* still collapses) |
| **2** | + JSON minify (whole-input *and* embedded pretty blocks), `\r` redraw collapse, ANSI strip, timestamps (ISO/syslog/nginx/HDFS/BGL/HealthApp), base64/base64url/hex blobs, UUIDs, venv paths, scattered-duplicate aliasing, template mining, similar-line collapse with value summaries | logs, CI dumps, traces, tool output | **the sweet spot and the default** — destroys machine noise, keeps every distinct fact |
| **2** | + JSON minify, `\r` redraw collapse, ANSI strip, ISO timestamps, base64/hex blobs, UUIDs, venv paths, scattered-duplicate aliasing, similar-line collapse | logs, CI dumps, traces, tool output | **the sweet spot** — destroys machine noise, keeps every distinct fact |
| **3** | + filler-phrase stripping ("could you please", "just") | prose, chat history | fine for text, never for strict logic |
| **4** | + `two_sticks` caveman mode: drops articles/copulas/auxiliaries, never negations or modals | gist-only prose, transcripts | lossy on style, protective of meaning — "do **not** delete" keeps its *not* |
"Keeps every distinct fact" is enforced, not hoped: when similar lines
collapse, the digits that varied are summarized in the marker —
`[3 similar lines omitted; values 404/429/503]` — because sometimes the
digits (status codes, ports, exit codes) *are* the diagnosis. An earlier
version silently ate them; the adversarial eval below is what caught it.
The crown jewels at level 2 are the two dictionary passes:
- `alias_repeats` — ordinary dedupe only sees *consecutive* repeats, so
interleaved multi-service logs sail straight through it. This pass hunts
scattered exact duplicates across the whole file and dictionary-codes them
(`@1 = ERROR [pool-3] psycopg2...` once in a legend, 2-token `@1`
everywhere else). Lossless — the legend keeps every line verbatim. On the
interleaved benchmark: 54,999 → 34,999 with dedupe alone (1.6x) vs
54,999 → 4,585 with aliasing (**12x**).
- `alias_templates` — real logs rarely repeat *exact* lines; they repeat
templates: `Failed password for root from 10.2.3.4 port 2201`. Two tiers of
deterministic, single-pass Drain-lite mining. Tier A masks hex/digit runs
*inside* tokens, so run-level constants inline into the template
(`@t1 = Failed password for root from 10.2.3.<*> port <*> ssh2`, per-line
refs `@t1 44 2201`). Tier B masks whole digit-bearing tokens on whatever
tier A left behind, so paths and ids that vary in letters still group —
which is what crushes compile/build logs. Each group is costed with exact
character accounting under both value encodings and takes the cheaper; a
template only fires when it provably pays. Every tier exists because a
public benchmark regressed without it: this pass took the LogHub corpus
from 1.8x to 3.9x.
The crown jewel at level 2 is `alias_repeats`: ordinary dedupe only sees
*consecutive* repeats, so interleaved multi-service logs sail straight through
it. `alias_repeats` hunts scattered duplicates across the whole file and
dictionary-codes them (`@1 = ERROR [pool-3] psycopg2...` once in a legend,
2-token `@1` everywhere else). It's lossless — the legend keeps every line
verbatim — and on the interleaved benchmark it's the difference between
54,999 → 34,999 (dedupe alone, 1.6x) and 54,999 → 4,585 (**12x**).
`--budget N` is the backstop, not the compressor: after the passes run, it
keeps head and tail lines and drops the middle with a `[~N tokens omitted]`
@ -132,155 +96,20 @@ context limit is a hard wall.
## The receipts
Every row reproducible from one command: `python3 bench.py` (o200k counts via
tiktoken — it refuses to print estimate numbers, so what you measure is what
this table claims).
All reproducible: `python3 bench.py` (o200k counts via tiktoken).
| Sample | Tokens | Why it wins |
|---|---|---|
| Mixed error log, 2000 lines | 86,914 → 6,730 (**12.9x**) | timestamp strip unmasks identical lines → dedupe; templates catch the numbered stragglers |
| Mixed error log, 2000 lines | 86,914 → 7,895 (**11.0x**) | timestamp strip unmasks identical lines → dedupe; similar-line collapse catches the numbered stragglers |
| Interleaved 3-service log, zero consecutive repeats | 54,999 → 4,585 (**12.0x**) | `alias_repeats` — consecutive dedupe alone managed 1.6x on this input |
| ANSI-colored CI log | 10,440 → 668 (**15.6x**) | color codes make identical lines look different; strip them and the log collapses |
| Captured pip/tqdm output | 10,465 → 216 (**98%**) | every overwritten `\r` progress frame is invisible on screen but real tokens in a capture |
| pytest failure dump | 732 → 515 (30%) | one site-packages path prefix = 25 tokens → 7 |
| Pretty-printed JSON API response | 8,932 → 3,234 (**64%**) | minify (round-trip verified lossless) + UUID→8-hex squash |
| ANSI-colored CI/docker build log | 21,198 → 2,387 (**8.9x**) | color codes make identical lines look different; strip them and the log collapses |
| Captured pip/tqdm output | 14,483 → 291 (**98%**) | every overwritten `\r` progress frame is invisible on screen but real tokens in a capture |
| pytest failure dump | 2,223 → 1,673 (25%) | one site-packages path prefix = 25 tokens → 7 |
| Pretty-printed JSON API response | 11,741 → 6,833 (42%) | minify (round-trip verified lossless) + UUID→8-hex squash |
| Chatty prose, level 4 | 427 → 264 (38%) | every dropped function word is a whole token; level 3 got 2% on the same text |
Throughput: measured **6–12 MB/s single-core** at level 2 (20 MB log in
2.6s). No model, no network — break-even input size is effectively zero.
## Real logs, real benchmarks
Synthetic logs flatter a compressor — they repeat exactly the way its passes
expect. So [bench_real.py](bench_real.py) runs three **public** corpora with
deterministic metrics (no model in the loop, no cherry-picking):
**[LogHub](https://github.com/logpai/loghub) samples** — 10 real system logs
(HDFS, BGL, OpenSSH, OpenStack...), 1.15M tokens: **3.9x overall**, ranging
from 2.0x (HDFS's unique block ids) to 9.8x (OpenStack). Fair warning made
explicit: real diverse logs compress 2–10x by structure, not the flat 13x of
repetitive error walls. It was this benchmark that forced `alias_templates`
and four new timestamp formats into existence — the corpus started at 1.5x.
**[LogChunks](https://zenodo.org/records/3632351)** — 797 real Travis CI
failure logs where *humans* marked the chunk that explains each failure
(externally validated with the developers who caused them). 66.2M tokens →
31.6M (**2.1x**), and **76.9% of labeled failure-explaining lines survive
byte-for-byte**; 67.5% of chunks survive fully intact. The rest is dominated
by similar-line collapse and templating, where the message survives once with
per-line values — represented, not deleted.
**[LogDx-CI](https://arxiv.org/abs/2605.28876)** — the benchmark built for
exactly this question: 35 real GitHub Actions failures with per-case
ground-truth diagnostic signals. All 35 cases, 15.0M tokens → 6.6M
(**2.3x**), with **95.2% of critical diagnostic signals retained** (139/146;
of the 7 misses, most are layout artifacts where templating separates a value
from its line). And the answer-quality check: on 12 cases, claude-haiku-4.5
diagnosed **12/12 from raw and 12/12 from compressed** — parity at ~2.4x
fewer tokens. (Ground-truth caveat inherited from the benchmark: its labels
are AI-drafted with single-author verification, n=35.) For calibration
against the field: rtk's most aggressive mode measures 810 tokens/case on
this same corpus at a 0.249 diagnosis score
([rtk#2012](https://github.com/rtk-ai/rtk/issues/2012)) — a far lossier
point on the curve. lessismore's stance is the opposite end: keep the
signals, then `--budget` down to whatever your cap is.
**The equal-budget receipt** — the question every harness actually faces is
not "compress or not" but "we cap tool output anyway; is compressing before
the cap worth it?" Measured on all 35 LogDx-CI cases (critical-signal
retention at the same token budget):
| Budget | truncate only | compress, then truncate |
|---|---|---|
| 2,000 tokens | 24.7% | **46.6%** |
| 8,000 tokens | 65.8% | **78.8%** |
| 32,000 tokens | 80.8% | **87.7%** |
One case in this run earned its keep twice: the benchmark caught a real
lessismore bug (a camelCase test name eaten as a base64 blob), which is now a
regression test. That's what public benchmarks are for.
A faithfulness note that cost real ratio: an earlier build scored "8.9x" on
LogHub because refs from *different* templates were collapsing into one
"similar lines" blob, silently dropping the middle lines' values. That
over-merge is now prevented (a ref's template name is part of its identity),
and the honest number is 3.9x. The same audit exposed that v0.3.0's HDFS
"win" was powered by the same phantom — this codebase prefers the smaller
true number.
## Does the model still get the answer?
A compression ratio proves the log got smaller, not that the signal
survived. So this is the eval that matters, and it's in the repo:
[eval.py](eval.py) generates eight seeded failure logs, each with one planted
root cause buried in machine noise **plus a red herring** (e.g. a cascade of
connection errors *caused by* an OOM kill 400 lines earlier), asks a model
"what's the root cause?" on the raw and the compressed version, and grades
the answer with a deterministic keyword check.
Result (claude-haiku-4.5, 2026-07-07, one run; this run went through the
Claude Code agent harness with the same single-turn prompts — `eval.py`'s
default runner is a logged-in `claude -p`. Rerun it yourself):
| Scenario | Raw tokens | Compressed | Raw | Compressed |
|---|---|---|---|---|
| OOM kill buried mid-log, connection-error red herring | 38,777 | 169 (**229x**) | ✗ blamed the red herring | ✓ found the OOM |
| 429 rate-limit burst hidden in a wall of status codes | 46,899 | 3,497 | ✗ "service overwhelmed" | ✓ named the 429s |
| Disk-full scattered through 3 interleaved services | 38,616 | 8,099 | ✓ | ✓ |
| One failed assertion in 1,400 ANSI test lines | 46,812 | 11,408 | ✓ | ✓ |
| Dependency conflict under progress-bar walls | 16,304 | 355 | ✓ | ✓ |
| Expired TLS cert in an nginx access-log wall | 56,696 | 11,855 | ✓ | ✓ |
| Missing env var in a crash-loop traceback | 7,680 | 1,610 | ✓ | ✓ |
| Java deadlock in a 40-thread dump | 8,157 | 3,742 (2.2x) | ✓ | ✓ |
| **Total** | **259,941** | **40,735 (6.4x)** | **6/8** | **8/8** |
The two raw failures are the lost-in-the-middle effect this tool exists to
counter: the model latched onto the loud symptom cascade and never surfaced
the quiet cause. Compressed, the cause is impossible to miss. The honest
caveats: n=1 run per condition, synthetic logs authored by this project, one
model — which is exactly why the harness ships in the repo with fixed seeds.
`python3 eval.py --dry` costs nothing and shows the scenarios; point
`LM_EVAL_CMD` at any prompt-on-stdin CLI to grade your own model.
## Where the tokens actually are: agent harnesses
A human remembering to pipe is the demo. The 50k-token walls of 2026 enter
context windows as **tool output inside agent harnesses**, so lessismore
ships two integrations:
**`lm --run`** — run the noisy command through the compressor and keep its
exit code (a shell pipe can't do that without pipefail games):
```bash
lm --run "pytest -x" # agent sees 500 tokens, not 20,000
lm --run "docker build ."
```
Tell your agent about it once (CLAUDE.md: *"run noisy commands through
`lm --run`"*) and every test run gets cheaper.
**`lm --hook`** — a Claude Code `PostToolUse` hook that compresses Bash
output before it enters the model's context. Zero dependencies, no `jq`;
outputs under 2,000 chars pass through untouched. In
`.claude/settings.json`:
```json
{
"hooks": {
"PostToolUse": [
{
"matcher": "Bash",
"hooks": [{ "type": "command", "command": "lm --hook" }]
}
]
}
}
```
Because the passes are deterministic, hook-compressed transcripts stay
byte-stable across re-runs — the prompt-cache argument above is strongest
exactly here. The hook fails open: on any input it doesn't recognize it
emits nothing, and Claude Code keeps the original output untouched.
Throughput: measured **~6 MB/s single-core** at level 2 (28 MB log in 4.4s).
No model, no network — break-even input size is effectively zero.
## What we refused to build (measured so it stays dead)
@ -305,14 +134,6 @@ against a real tokenizer:
digits), **separator shortening** (a 78-char `----` is already 1 token),
**prefix hoisting** (one stray line kills it), **JSON→TSV / float
truncation / pointer squash** (real savings, worse trade).
- **Drain-style letter-varying template merging** (clustering templates that
differ in ≤N word positions, so usernames/hostnames join one template) —
measured on LogHub: 3.0x → 3.0x, with Linux *regressing* 6.4x → 5.3x. The
extra per-line wildcard values eat exactly what the merged legends save.
- **Cross-template ref collapse** — letting `dedupe_similar` merge refs from
different templates scored 8.9x on LogHub, but the "compression" was
silently deleting the middle lines' values. Removed on faithfulness
grounds; the ratio was a lie.
## Battle-tested
@ -322,11 +143,7 @@ db" → "was **not** the db"), URLs eaten as base64, crashes on empty and
non-UTF-8 stdin, distinct SHA-256s falsely merging as "repeated",
`two_sticks` eating "IT" and "US" as function words. Nine fixed with
regression tests, four documented below, one wontfix (adversarial in-band
marker collision). A second review round found the big one: similar-line
collapse silently eating distinct status codes — fixed with value summaries
and now guarded by the downstream eval. A third bug (camelCase test names
eaten as base64 blobs) was caught by the LogDx-CI benchmark rather than a
reviewer. `python3 test_lessismore.py` — 48 asserts, no framework.
marker collision). `python3 test_lessismore.py` — 30 asserts, no framework.
## Where it does nothing (on purpose)
@ -335,56 +152,26 @@ and machine noise; it cannot compress information, and doesn't pretend to.
| Content | Expect | Verdict |
|---|---|---|
| Repetitive machine output | 5–16x, up to ~230x on pathological repeats | the reason this exists |
| Real-world CI / system logs | 2–4x typical, up to 10x on template-heavy logs (measured on LogHub, LogChunks, LogDx-CI) | the honest production number |
| Structured data (JSON, tracebacks) | 30–64% | worthwhile, lossless where it fires |
| Repetitive machine output | 5–12x, up to 50x on pathological repeats | the reason this exists |
| Structured data (JSON, tracebacks) | 25–42% | worthwhile, lossless where it fires |
| Varied prose | ~2% (level 3) / 38% lossy (level 4) | gist mode only |
| Clean code, unique dense text, thread dumps | ~0% by design | that's what `--ml` or truncation is for |
| Clean code, unique dense text | ~0% by design | that's what `--ml` or truncation is for |
`--ml RATE` bolts on Microsoft's LLMLingua-2 for perplexity pruning as a
last mile (`pip install llmlingua`) — runs *after* the deterministic passes so
you're not paying a classifier model to delete duplicate log lines. Only
worth it on multi-KB inputs, and it forfeits the cache-stability guarantee.
## Related work (know the river you're panning)
Honest placement, so you can pick the right tool:
- **[rtk](https://github.com/rtk-ai/rtk)** — Rust proxy with per-command
filters (ANSI strip, dedupe, failures-only test output), hooks into agent
CLIs. Command-aware and aggressive where lessismore is generic text passes
and lossless-leaning: rtk decides what you need to see; lessismore removes
only what is provably redundant and keeps a legend.
- **[Headroom](https://github.com/chopratejas/headroom)** — the heavyweight
tool-output compressor (Python, proxy + MCP). Has ML models in the loop,
so it needs a cache-alignment component to patch the non-determinism
lessismore doesn't have.
- **[Drain3](https://github.com/logpai/Drain3)** — industrial log template
mining. `alias_templates` is a deterministic, single-pass, zero-dep
Drain-lite (two-tier hex-run/whole-token masking with exact payoff
accounting, instead of similarity clustering); if you need streaming
template state across files, or one template spanning letter-varying
parameters (we measured that merge and it didn't pay — see the graveyard),
use Drain3.
- **[LLMLingua-2](https://github.com/microsoft/LLMLingua)** — ML token
pruning; wrapped here as the optional `--ml` last mile, *after* the
deterministic passes, so you never pay a classifier to delete duplicate
log lines.
- **[Dictionary-encoding prompt compression](https://arxiv.org/abs/2604.13066)**
(2026) — independently validates the `@1 = line` legend technique
academically (≥0.99 fidelity on log benchmarks); no code released.
lessismore is, in effect, a reference implementation.
What none of them package together — and the reason this exists — is the
combination: pure-stdlib, deterministic end to end, tokenizer-measured, with
the cache-stability argument as a design constraint rather than a patch.
## Try it in a browser
`lm --serve` runs a paste-in demo page on `http://localhost:7777` — paste your
ugliest log, pick a level, watch the token count drop. Stdlib only, binds
localhost only.
## License
MIT — see [LICENSE](LICENSE).
## Known tradeoffs
- Levels 2+ assume line *order* matters but wall-clock timing doesn't. When
@ -394,24 +181,4 @@ localhost only.
- At levels 3–4, dedupe counts describe the post-stripped text — five
differently-phrased "retry the job" lines can legitimately merge.
- In-band markers can collide with input that already contains them;
`alias_repeats` and `alias_templates` bail out if their own `@N`/`@tN`
markers already appear as lines.
- Blob squashing eats base64url runs (real JWTs) only when they look like
blobs (digits + mixed case, no 16+ letter run); a kebab-case slug or a
camelCase test name is treated as content and survives.
- Similar-line collapse summarizes varying digits per column
(`values 404/429/503`, or a min–max range past 4 distinct); if the digit
*runs* per line differ in count, it falls back to a plain count marker.
- `alias_templates` reconstructs lines by filling `<*>` left to right with
the per-line values — lossless modulo whitespace runs (a templated line's
original spacing/alignment is not preserved). Parameters that vary in
*letters* (usernames, hostnames) split into separate templates rather than
merging — less compression, never confusion.
- Reading a templated log requires joining refs back to the legend. Models
handle this fine (the evals above are all on templated output, and answers
routinely cite `@N` legend entries), but a human skimming the compressed
file will prefer the raw one.
## License
MIT — see [LICENSE](LICENSE).
`alias_repeats` bails out if its own `@N` markers already appear as lines.

64
bench.py
View File

@ -1,19 +1,7 @@
"""Reproduce the README benchmark numbers — every row of the receipts table.
Requires tiktoken (pip install tiktoken): the README numbers are o200k counts,
and printing chars/4 estimates here would let the receipts silently drift from
the claims. Pass --estimate if you really want the guesses.
"""
import json
"""Reproduce the README benchmark numbers. pip install tiktoken for exact counts."""
import random
import sys
from lessismore import compress, count_tokens, _encoder
if _encoder() is None and "--estimate" not in sys.argv:
sys.exit("bench.py: tiktoken not installed — these would be chars/4 GUESSES,\n"
"not the o200k-measured numbers the README claims.\n"
" pip install tiktoken (or pass --estimate to see the guesses anyway)")
from lessismore import compress, count_tokens
def bench(name, text, level):
@ -35,6 +23,7 @@ lines.insert(500, "2026-07-07T10:08:20.123Z DEBUG auth token=" + "eyJhbGciOiJIUz
bench("mixed error log", "\n".join(lines), 2)
# interleaved log: three services alternating — zero consecutive runs, dedupe-proof
random.seed(2)
msgs = ["ERROR [pool-3] psycopg2.OperationalError: connection to server at "
"10.0.0.5 failed: Connection refused",
"WARN [redis-1] retry queue depth exceeded threshold, backing off 500ms",
@ -42,53 +31,6 @@ msgs = ["ERROR [pool-3] psycopg2.OperationalError: connection to server at "
bench("interleaved log", "\n".join(
f"2026-07-07T10:{i // 60 % 60:02d}:{i % 60:02d}Z " + msgs[i % 3] for i in range(1500)), 2)
# ANSI CI log: color codes make identical lines look distinct to any dedupe
random.seed(3)
ci = []
for step in range(1, 13):
ci.append(f"\x1b[1m\x1b[34mStep {step}/12\x1b[0m : RUN pip install -r requirements.txt")
for _ in range(random.randint(20, 60)):
ci.append("\x1b[36m ---> Using cache\x1b[0m")
for pkg in range(random.randint(5, 15)):
ci.append(f"\x1b[90m2026-07-07T10:0{step % 10}:00Z\x1b[0m \x1b[32m✓\x1b[0m "
f"Collecting dependency {pkg} (cached wheel, {random.randint(10, 900)} kB)")
bench("ANSI CI log", "\n".join(ci), 2)
# captured pip/tqdm: hundreds of \r-overwritten progress frames, invisible on screen
frames = []
for pkg in ("requests", "urllib3", "charset_normalizer", "idna", "certifi"):
frames.append(f"Collecting {pkg}\n Downloading {pkg}-2.0.0-py3-none-any.whl (150 kB)\n")
frames.append("".join(f"\r |{'█' * (i // 4)}{' ' * (25 - i // 4)}| "
f"{i}% {i * 15 // 10} kB {random.randint(100, 999)} kB/s"
for i in range(1, 101)) + "\n")
bench("pip/tqdm capture", "".join(frames) + "Successfully installed requests-2.0.0\n", 2)
# pytest failure dump: one venv path prefix repeated across every traceback frame
venv = "/Users/dev/project/.venv/lib/python3.12/site-packages"
tb = ["=================================== FAILURES ==================================",
"_________________________________ test_checkout _______________________________"]
for mod, ln in [("httpx/_client", 1054), ("httpx/_transports/default", 249),
("httpcore/_sync/connection_pool", 216), ("httpcore/_sync/connection", 99),
("httpcore/_sync/http11", 136), ("httpcore/_backends/sync", 126)] * 3:
tb += [f' File "{venv}/{mod}.py", line {ln}, in request',
" raise exc from None"]
tb += ["E httpcore.ConnectTimeout: timed out",
"=========================== short test summary info ==========================",
"FAILED tests/test_checkout.py::test_checkout - httpcore.ConnectTimeout"]
bench("pytest failure dump", "\n".join(tb), 2)
# pretty-printed JSON API response: indentation + UUIDs are the tax
random.seed(4)
resp = {"data": [{"id": f"{random.getrandbits(32):08x}-{random.getrandbits(16):04x}-"
f"4{random.getrandbits(12):03x}-a{random.getrandbits(12):03x}-"
f"{random.getrandbits(48):012x}",
"status": random.choice(["active", "pending"]),
"amount": random.randint(100, 99999),
"currency": "USD",
"created_at": "2026-07-07T10:00:00Z"} for _ in range(120)],
"page": 1, "total": 120}
bench("pretty JSON response", json.dumps(resp, indent=2), 2)
# chatty prose: levels 3 vs 4 show where filler stripping ends and caveman starts
prose = ("So basically what happened was that the team had been trying to get the "
"deployment pipeline working for about three weeks, and it turned out that the "

142
bench_real.py
View File

@ -1,142 +0,0 @@
"""bench_real.py — measure lessismore on REAL public log corpora, not synthetic ones.
Three benchmarks, all free, all with a deterministic metric (no model calls):
loghub 10 LogHub sample logs (HDFS, BGL, OpenSSH, ...) — token ratios on
real system logs. Downloads ~2.8 MB from GitHub on first run.
https://github.com/logpai/loghub (CC-BY-4.0)
logchunks 797 Travis CI failure logs with HUMAN-labeled failure-explaining
chunks (MSR 2020). Metric: do the labeled chunk lines survive?
curl -L -o LogChunks.zip \\
"https://zenodo.org/api/records/3632351/files/LogChunks.zip/content"
unzip LogChunks.zip
python3 bench_real.py logchunks ./LogChunks
logdx LogDx-CI: 35 real GitHub Actions failures with per-case
required-signal ground truth (arXiv 2605.28876, CC-BY-4.0).
Metric: are the critical diagnostic signals still present?
pip install huggingface_hub
hf download eyuansu71/logdx-ci --repo-type dataset --local-dir ./logdx-ci
python3 bench_real.py logdx ./logdx-ci
Chunk/signal matching normalizes whitespace and ANSI on both sides (the labels
themselves contain terminal color remnants), nothing else.
"""
import json
import re
import sys
import urllib.request
import xml.etree.ElementTree as ET
from pathlib import Path
from lessismore import budget, compress, count_tokens
_CSI = re.compile(r"\x1b?\[[0-9;]*[A-Za-z]") # ANSI with or without the ESC byte
def norm(s):
return re.sub(r"\s+", " ", _CSI.sub("", s)).strip()
LOGHUB = ["HDFS", "BGL", "OpenStack", "Zookeeper", "Apache",
"OpenSSH", "Spark", "Thunderbird", "Linux", "HealthApp"]
def bench_loghub(cache=Path("loghub_samples")):
cache.mkdir(exist_ok=True)
tot_b = tot_a = 0
print(f"{'dataset':14} {'raw tok':>8} {'small':>8} {'ratio':>6}")
for name in LOGHUB:
f = cache / f"{name}_2k.log"
if not f.exists():
url = f"https://raw.githubusercontent.com/logpai/loghub/master/{name}/{name}_2k.log"
f.write_bytes(urllib.request.urlopen(url, timeout=60).read())
text = f.read_text(encoding="utf-8", errors="replace")
b, a = count_tokens(text), count_tokens(compress(text, 2))
tot_b += b; tot_a += a
print(f"{name:14} {b:>8,} {a:>8,} {b/a:>5.1f}x")
print(f"{'TOTAL':14} {tot_b:>8,} {tot_a:>8,} {tot_b/tot_a:>5.1f}x")
def bench_logchunks(root):
root = Path(root)
tot_b = tot_a = n = lv = la = 0
full_v = 0
for xf in sorted(root.glob("build-failure-reason/*/*.xml")):
try:
tree = ET.parse(xf)
except ET.ParseError:
continue
for ex in tree.getroot().iter("Example"):
p = root / "logs" / ex.findtext("Log", "").strip()
chunk = ex.findtext("Chunk", "")
if not (p.exists() and chunk.strip()):
continue
out = compress(p.read_text(encoding="utf-8", errors="replace"), 2)
tot_b += count_tokens(p.read_text(encoding="utf-8", errors="replace"))
tot_a += count_tokens(out)
n += 1
nout = norm(out)
clines = [norm(l) for l in chunk.split("\n") if norm(l)]
v = sum(1 for l in clines if l in nout)
lv += v; la += len(clines); full_v += v == len(clines)
print(f"cases: {n} tokens {tot_b:,} -> {tot_a:,} ({tot_b/tot_a:.1f}x, {1-tot_a/tot_b:.0%} saved)")
print(f"labeled chunk lines retained verbatim: {lv}/{la} ({lv/la:.1%})")
print(f"chunks fully verbatim: {full_v}/{n} ({full_v/n:.1%})")
print("(the rest is mostly similar-line collapse: the message survives once "
"with a value summary — see README)")
def bench_logdx(root):
root = Path(root)
gts = sorted(set(root.glob("cases/**/ground_truth.json")))
caps = (2000, 8000, 32000)
tot_b = tot_a = ck = ca = 0
cap_raw = {c: 0 for c in caps} # truncate-only retention
cap_cmp = {c: 0 for c in caps} # compress-then-truncate retention
for gt_path in gts:
raw_p = gt_path.parent / "raw.log"
if not raw_p.exists():
continue
raw = raw_p.read_text(encoding="utf-8", errors="replace")
out = compress(raw, 2)
tot_b += count_tokens(raw); tot_a += count_tokens(out)
nout = norm(out)
ntr = {c: norm(budget(raw, c)) for c in caps}
ncm = {c: norm(budget(out, c)) for c in caps}
for sig in json.loads(gt_path.read_text()).get("required_signals", []):
if sig.get("importance") != "critical":
continue
texts = [norm(t) for t in [sig.get("value"), sig.get("file")]
+ sig.get("aliases", []) if t]
ca += 1
ck += any(t in nout for t in texts)
for c in caps:
cap_raw[c] += any(t in ntr[c] for t in texts)
cap_cmp[c] += any(t in ncm[c] for t in texts)
print(f"cases: {len(gts)} tokens {tot_b:,} -> {tot_a:,} ({tot_b/tot_a:.1f}x, {1-tot_a/tot_b:.0%} saved)")
print(f"critical diagnostic signals retained: {ck}/{ca} ({ck/ca:.1%})")
print("equal-budget (the question a harness actually faces):")
for c in caps:
print(f" {c:>6,} tokens: truncate-only {cap_raw[c]/ca:.1%} "
f"compress-then-truncate {cap_cmp[c]/ca:.1%}")
def main():
if len(sys.argv) < 2 or sys.argv[1] not in ("loghub", "logchunks", "logdx"):
sys.exit(__doc__)
if sys.argv[1] == "loghub":
bench_loghub()
elif sys.argv[1] == "logchunks":
if len(sys.argv) < 3:
sys.exit("usage: bench_real.py logchunks <path-to-extracted-LogChunks>")
bench_logchunks(sys.argv[2])
else:
if len(sys.argv) < 3:
sys.exit("usage: bench_real.py logdx <path-to-logdx-ci-dataset>")
bench_logdx(sys.argv[2])
if __name__ == "__main__":
main()

362
build/lib/lessismore.py Normal file
View File

@ -0,0 +1,362 @@
"""lessismore — squeeze text before it hits an LLM.
Deterministic passes first (free, safe, cacheable). Optional ML pruning
(LLMLingua-2) only when installed and only worth it on big inputs.
from lessismore import compress
small = compress(big_log, level=2)
$ python3 lessismore.py dump.log -l 2 > small.log
$ tail -5000 app.log | python3 lessismore.py -l 2 | llm ...
"""
import json
import re
import sys
from collections import Counter
from functools import lru_cache
# ---------------------------------------------------------------- tokens
@lru_cache(maxsize=1)
def _encoder():
try:
import tiktoken
return tiktoken.get_encoding("o200k_base")
except ImportError:
return None
def count_tokens(text: str) -> int:
enc = _encoder()
if enc:
return len(enc.encode(text))
return max(1, len(text) // 4) # ponytail: chars/4 heuristic; pip install tiktoken for real counts
# ---------------------------------------------------------------- passes
# Each pass is (str) -> str. Order matters: whitespace before dedupe.
def collapse_whitespace(text: str) -> str:
text = re.sub(r"[ \t]+$", "", text, flags=re.M) # trailing whitespace
text = re.sub(r"(?<=\S)[ \t]{2,}", " ", text) # interior runs (leading indent kept: code-safe)
text = re.sub(r"\n{3,}", "\n\n", text) # blank-line runs
return text
def dedupe_lines(text: str) -> str:
"""Collapse runs of identical lines — the classic log killer."""
lines = text.split("\n")
out, i = [], 0
while i < len(lines):
j = i
while j < len(lines) and lines[j] == lines[i]:
j += 1
run = j - i
# marker must actually be shorter than the lines it replaces
if run >= 4 and lines[i].strip() and (run - 1) * (len(lines[i]) + 1) > 45:
out.append(lines[i])
out.append(f"[previous line repeated {run - 1} more times]")
else:
out.extend(lines[i:j])
i = j
# ponytail: only consecutive repeats; add block-level dedupe (repeated stack traces) if logs demand it
return "\n".join(out)
# no '/' in the class: URL paths are ≥64-char alnum+slash runs and they ARE the content
_BLOB = re.compile(r"\b(?:[A-Za-z0-9+]{64,}={0,2}|[0-9a-fA-F]{48,})\b")
def squash_blobs(text: str) -> str:
"""Base64/hex runs are token-dense and semantically opaque — keep head and tail.
The tail suffix keeps two different blobs from squashing to the same stub
and then being falsely merged as "repeated" by dedupe_lines.
"""
return _BLOB.sub(lambda m: f"{m.group()[:12]}[+{len(m.group()) - 16} chars]{m.group()[-4:]}", text)
_TIMESTAMP = re.compile(
r"\b\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}(?:[.,]\d+)?(?:Z|[+-]\d{2}:?\d{2})?\b ?"
)
def strip_timestamps(text: str) -> str:
"""Line order already encodes sequence; per-line ISO timestamps are ~8 tokens each."""
# ponytail: ISO-8601 only; add syslog/other formats when a real log needs them
return _TIMESTAMP.sub("", text)
def minify_json(text: str) -> str:
"""Whole-input pretty JSON → minified. Lossless when it fires, untouched when not."""
try:
return json.dumps(json.loads(text), separators=(",", ":"), ensure_ascii=False)
except ValueError:
return text
def collapse_cr(text: str) -> str:
"""Keep only the final state of \\r-overwritten progress lines (pip/tqdm/wget)."""
text = text.replace("\r\n", "\n")
return "\n".join(l.rsplit("\r", 1)[-1] for l in text.split("\n"))
_ANSI = re.compile(r"\x1b\[[0-9;?]*[A-Za-z]") # ponytail: CSI only; add OSC if titles show up
def strip_ansi(text: str) -> str:
"""Color codes carry nothing for an LLM, and they make identical lines differ."""
return _ANSI.sub("", text)
_UUID = re.compile(r"\b[0-9a-fA-F]{8}-(?:[0-9a-fA-F]{4}-){3}[0-9a-fA-F]{12}\b")
def squash_uuids(text: str) -> str:
"""36 chars → 8-hex prefix, git-short-hash style; cross-references still resolve."""
return _UUID.sub(lambda m: m.group()[:8] + "…", text)
_PKGPATH = re.compile(r'[^\s"\']+/(?:site-packages|dist-packages|lib/python3\.\d+)/')
def squash_pkgpaths(text: str) -> str:
"""Traceback path spam: …/httpx/_client.py:1054 is still unique without the venv prefix."""
return _PKGPATH.sub("…/", text)
# the lookbehinds keep "not just X" from becoming "not X" — meaning inversion
_FILLER = re.compile(
r"(?<!\bnot )(?<!n't )"
r"\b(?:could you please|can you please|i would like you to|i want you to|"
r"go ahead and|hey there,?|please|kindly|basically|actually|currently|"
r"really|simply|just|very|quite)\b ?",
re.I,
)
def strip_filler(text: str) -> str:
"""Aggressive: eats words like 'just' anywhere, including inside strings. Prose only."""
return _FILLER.sub("", text)
# caveman-style word dropping: every function word is a whole token.
# NEVER add negations (not/no/never), modals (must/should), or order words
# (before/after) — dropping those changes meaning, not just style.
_STICKS = re.compile(
r"\b(?:the|a|an|is|are|was|were|be|been|being|am|i|we|you|they|it|"
r"that|which|who|have|has|had|do|does|did|there)\b ?",
re.I,
)
def two_sticks(text: str) -> str:
"""Gist-only: strips text to caveman. Fine for articles/transcripts, never instructions."""
# keep IT/US-style acronyms that case-insensitively collide with function words
return _STICKS.sub(lambda m: m.group() if m.group().strip().isupper()
and len(m.group().strip()) > 1 else "", text)
_SKEL = re.compile(r"[^A-Za-z]+")
def dedupe_similar(text: str) -> str:
"""Collapse runs of lines identical after masking non-letters — 'same words,
different numbers' (progress lines, per-item CI steps). Keeps first and last
so progression endpoints survive."""
lines = text.split("\n")
out, i = [], 0
while i < len(lines):
k, j = _SKEL.sub(" ", lines[i]).strip(), i
while j < len(lines) and _SKEL.sub(" ", lines[j]).strip() == k:
j += 1
if j - i >= 4 and k:
out += [lines[i], f"[{j - i - 2} similar lines omitted]", lines[j - 1]]
else:
out.extend(lines[i:j])
i = j
return "\n".join(out)
_REF = re.compile(r"@\d+")
def alias_repeats(text: str, min_count: int = 4, min_len: int = 30) -> str:
"""Dictionary-code scattered duplicate lines that consecutive dedupe can't reach.
Lossless — the legend keeps every line verbatim. A code only pays when it
replaces a repeated multi-token sequence; single words are already 1 BPE
token each, so word-level codebooks lose (measured, see README).
"""
lines = text.split("\n")
if any(_REF.fullmatch(l) for l in lines): # already aliased, or real @N content — bail
return text
counts = Counter(l for l in lines if len(l) >= min_len)
# ponytail: c*len>200 chars is the payoff heuristic; tune if legends ever dominate
worth = [l for l, c in counts.items() if c >= min_count and c * len(l) > 200]
if not worth:
return text
ref = {l: f"@{i}" for i, l in enumerate(worth, 1)}
legend = [f"@{i} = {l}" for i, l in enumerate(worth, 1)]
return "\n".join(["[repeated lines aliased:]"] + legend + [""] +
[ref.get(l, l) for l in lines])
# order: strippers leave doubled spaces, so collapse_whitespace runs after them;
# normalized lines then match better in alias_repeats/dedupe_lines/dedupe_similar
_STRIP2 = [minify_json, collapse_cr, strip_ansi, strip_timestamps,
squash_blobs, squash_uuids, squash_pkgpaths]
_MERGE = [collapse_whitespace, alias_repeats, dedupe_lines, dedupe_similar]
LEVELS = {
1: [collapse_whitespace, dedupe_lines], # code-safe
2: _STRIP2 + _MERGE, # logs/dumps/tool output
3: _STRIP2 + [strip_filler] + _MERGE, # prose
4: _STRIP2 + [strip_filler, two_sticks] + _MERGE, # gist-only
}
def compress(text: str, level: int = 1) -> str:
for f in LEVELS[level]:
text = f(text)
return text
def budget(text: str, max_tokens: int) -> str:
"""Hard cap: keep head and tail lines, drop the middle with a marker.
The only pass with GUARANTEED bounded output — run it after compression,
as the backstop, never instead of it.
"""
total = count_tokens(text)
if total <= max_tokens:
return text
lines = text.split("\n")
limit = max(0, max_tokens - 8) # reserve for the marker line
head, tail = [], []
h_tok = t_tok = 0
hi, ti = 0, len(lines) - 1
while hi <= ti:
take_head = h_tok <= t_tok
line = lines[hi] if take_head else lines[ti]
tok = count_tokens(line) + 1
if h_tok + t_tok + tok > limit:
break
if take_head:
head.append(line); h_tok += tok; hi += 1
else:
tail.append(line); t_tok += tok; ti -= 1
if not head and not tail: # one giant line (e.g. minified JSON): slice by chars
keep = limit * 2 # ~4 chars/token, half per side
head, tail = [text[:keep]], [text[-keep:]]
h_tok = count_tokens(head[0]); t_tok = count_tokens(tail[0])
omitted = max(0, total - h_tok - t_tok)
return "\n".join(head + ["[~%d tokens omitted]" % omitted] + list(reversed(tail)))
def serve(port=7777):
"""Paste-in demo page on localhost. Zero deps, binds 127.0.0.1 only."""
import html
import http.server
import urllib.parse
page = ("<!doctype html><meta charset=utf-8><title>lessismore</title>"
"<style>body{font-family:system-ui;max-width:900px;margin:2rem auto;padding:0 1rem;"
"background:#10141a;color:#e6edf3}textarea{width:100%;height:35vh;font-family:monospace;"
"background:#1a2028;color:#e6edf3;border:1px solid #2d3743;border-radius:8px;padding:8px}"
"select,button{font-size:16px;padding:8px 14px;border-radius:8px;border:1px solid #2d3743;"
"background:#1f6feb;color:#fff;cursor:pointer}select{background:#1a2028}</style>"
"<h1>lessismore \U0001F4C9</h1>"
"<form method=post><textarea name=text placeholder='paste your ugliest log here'></textarea>"
"<p>level <select name=level><option>1<option selected>2<option>3<option>4</select> "
"<button>squeeze</button></p></form>{result}")
class Handler(http.server.BaseHTTPRequestHandler):
def log_message(self, *args):
pass
def _page(self, result=""):
body = page.replace("{result}", result).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def do_GET(self):
self._page()
def do_POST(self):
n = int(self.headers.get("Content-Length", 0))
q = urllib.parse.parse_qs(self.rfile.read(n).decode("utf-8", "replace"))
text = q.get("text", [""])[0]
level = min(4, max(1, int(q.get("level", ["2"])[0])))
out = compress(text, level)
b, a = count_tokens(text), count_tokens(out)
self._page("<p><b>{:,} → {:,} tokens ({}% saved)</b></p>"
"<textarea readonly>{}</textarea>".format(
b, a, round(100 * (1 - a / max(b, 1))), html.escape(out)))
print("lessismore UI on http://localhost:%d (Ctrl+C to stop)" % port)
http.server.ThreadingHTTPServer(("127.0.0.1", port), Handler).serve_forever()
# ---------------------------------------------------------------- optional ML pass
@lru_cache(maxsize=1)
def _llmlingua():
from llmlingua import PromptCompressor # pip install llmlingua
return PromptCompressor(
model_name="microsoft/llmlingua-2-xlm-roberta-large-meetingbank",
use_llmlingua2=True,
)
def compress_ml(text: str, rate: float = 0.5) -> str:
"""Perplexity-based token pruning (LLMLingua-2).
Runs a local classifier model — only pays for itself on multi-KB inputs.
Run the deterministic passes first; never feed it code you need verbatim.
"""
return _llmlingua().compress_prompt(text, rate=rate)["compressed_prompt"]
# ---------------------------------------------------------------- CLI
def main():
import argparse
p = argparse.ArgumentParser(prog="lessismore", description=__doc__.splitlines()[0])
p.add_argument("file", nargs="?", help="input file (default: stdin)")
p.add_argument("-l", "--level", type=int, default=1, choices=sorted(LEVELS),
help="1=code-safe 2=logs/dumps 3=prose 4=gist-only caveman (default 1)")
p.add_argument("--ml", type=float, metavar="RATE",
help="also run LLMLingua-2 keeping RATE of tokens (needs: pip install llmlingua)")
p.add_argument("--budget", type=int, metavar="N",
help="hard cap output at ~N tokens: keep head+tail, drop the middle")
p.add_argument("--serve", nargs="?", const=7777, type=int, metavar="PORT",
help="serve a paste-in demo page on localhost (default port 7777)")
a = p.parse_args()
if a.serve:
return serve(a.serve)
# newline="" / buffer.read(): keep \r intact for collapse_cr
raw = (open(a.file, encoding="utf-8", errors="replace", newline="").read() if a.file
else sys.stdin.buffer.read().decode("utf-8", "replace"))
out = compress(raw, a.level)
if a.ml:
try:
out = compress_ml(out, a.ml)
except ImportError:
sys.exit("--ml needs: pip install llmlingua")
if a.budget:
out = budget(out, a.budget)
sys.stdout.write(out)
before, after = count_tokens(raw), count_tokens(out)
print(f"lessismore: {before} → {after} tokens ({1 - after / max(before, 1):.0%} saved)",
file=sys.stderr)
if __name__ == "__main__":
main()

236
eval.py
View File

@ -1,236 +0,0 @@
"""eval.py — does the model still get the answer after compression?
Compression ratios prove the log got smaller, not that the signal survived.
This is the test that matters: eight synthetic-but-realistic failure logs,
each with ONE planted root cause buried in machine noise (plus red herrings),
asked to a model twice — raw vs `compress(level=2)` — and graded by a
deterministic keyword check on the answer. If compression eats the diagnosis,
this catches it.
python3 eval.py --dry # build scenarios, show token counts, no model
python3 eval.py # ask the model (needs a logged-in `claude` CLI)
LM_EVAL_CMD="llm -m gpt-5-mini" python3 eval.py # any prompt-on-stdin CLI
Scenario generation is seeded — same logs every run. Model answers vary run
to run; the binary keyword grading absorbs phrasing differences.
"""
import argparse
import os
import random
import re
import shlex
import subprocess
import sys
from lessismore import compress, count_tokens
QUESTION = ("You are debugging a production incident. Above is the captured "
"log output. In 1-2 sentences: what is the ROOT CAUSE of the failure?")
MODEL_CMD = os.environ.get("LM_EVAL_CMD", "claude -p --model claude-haiku-4-5-20251001")
# ---------------------------------------------------------------- scenarios
# Each returns (log_text, expect_regex). The regex is what a correct root-cause
# answer must mention; red herrings are planted so a wrong answer fails it.
def s_buried_oom():
"""Kernel OOM-kills postgres mid-log; the tail is 400 lines of red-herring
connection errors that a lost-in-the-middle model blames instead."""
r, out = random.Random(11), []
for i in range(700):
out.append(f"2026-07-07T09:{i // 60:02d}:{i % 60:02d}Z INFO [api] "
f"GET /v1/orders/{r.randint(1000, 9999)} 200 in {r.randint(8, 90)}ms")
out.append("2026-07-07T09:11:40Z kernel: Out of memory: Killed process 2211 (postgres) "
"total-vm:8123456kB, anon-rss:7901234kB")
out.append("2026-07-07T09:11:40Z postgres[2211]: FATAL: terminating connection due to "
"unexpected postmaster exit")
for i in range(400):
out.append(f"2026-07-07T09:{12 + i // 60:02d}:{i % 60:02d}Z ERROR [api] "
f"psycopg2.OperationalError: connection to server at 10.0.0.5 refused")
return "\n".join(out), r"out of memory|oom|killed process|memory"
def s_status_codes():
"""Rate limiting (429) from the payments API triggers cascading 500s. The
codes differ only in digits — exactly what a careless similar-line collapse
would eat."""
r, out = random.Random(12), []
for i in range(1400):
if i < 600:
c, ms = 200, r.randint(40, 200)
elif i < 700:
c, ms = r.choice([429, 429, 429, 200]), r.randint(5, 30)
else:
c, ms = r.choice([500, 502, 500]), r.randint(2000, 3100)
out.append(f"2026-07-07T14:{i // 60 % 60:02d}:{i % 60:02d}Z gateway: upstream "
f"payments-api returned status {c} for POST /api/checkout in {ms}ms")
return "\n".join(out), r"429|rate.?limit|too many requests"
def s_interleaved_disk():
"""Three services interleave (zero consecutive repeats — dedupe-proof);
the db's 'No space left on device' is scattered 1-in-40."""
r, out = random.Random(13), []
for i in range(1500):
ts = f"2026-07-07T16:{i // 60 % 60:02d}:{i % 60:02d}Z"
which = i % 3
if which == 0:
out.append(f"{ts} INFO [api] request {r.randint(10000, 99999)} completed")
elif which == 1:
out.append(f"{ts} WARN [worker] job retry {r.randint(1, 5)} scheduled, backing off")
elif i % 40 == 2:
out.append(f"{ts} ERROR [db] could not extend file base/16384/2619: "
f"No space left on device")
else:
out.append(f"{ts} INFO [db] checkpoint complete: wrote {r.randint(100, 999)} buffers")
return "\n".join(out), r"space|disk|storage|full"
def s_ansi_ci():
"""1400 green PASSED lines in full ANSI dress; one red FAILED assertion
names the offending function."""
r, out = random.Random(14), []
mods = ["auth", "cart", "checkout", "billing", "search", "profile"]
for i in range(1400):
m = r.choice(mods)
out.append(f"\x1b[32mPASSED\x1b[0m tests/test_{m}.py::test_{m}_{r.randint(1, 99):02d} "
f"\x1b[90m({r.randint(1, 40)}ms)\x1b[0m")
if i == 981:
out.append("\x1b[31mFAILED\x1b[0m tests/test_billing.py::test_invoice_total")
out.append("\x1b[31mE AssertionError: round_half(2.675) == 2.68, got 2.67 — "
"float truncation in round_half()\x1b[0m")
out.append("\x1b[31m1 failed\x1b[0m, \x1b[32m1401 passed\x1b[0m in 42.31s")
return "\n".join(out), r"round_half|rounding|truncat|2\.6[78]"
def s_pip_conflict():
"""Progress-bar walls (\\r frames) drown a one-line dependency conflict."""
r, out = random.Random(15), []
for pkg in ("numpy", "pandas", "scipy", "matplotlib", "scikit_learn", "torch"):
out.append(f"Collecting {pkg}")
out.append(f" Downloading {pkg}-2.1.0-cp312-cp312-macosx_11_0_arm64.whl "
f"({r.randint(1, 80)}.{r.randint(0, 9)} MB)")
out.append("".join(f"\r |{'█' * (i // 3)}{' ' * (34 - i // 3)}| {i}% "
f"{r.randint(100, 999)}.{r.randint(0, 9)} kB/s eta 0:00:{99 - i:02d}"
for i in range(1, 101)))
out.append("ERROR: Cannot install app 1.0 because requests 2.32.0 requires urllib3<3, "
"but you have urllib3 3.0.1 which is incompatible.")
return "\n".join(out), r"urllib3"
def s_cert_expired():
"""An nginx access-log wall; the error-log lines that matter say the
upstream's TLS certificate expired."""
r, out = random.Random(16), []
for i in range(1300):
ts = f"[07/Jul/2026:18:{i // 60 % 60:02d}:{i % 60:02d} +0000]"
if i % 60 == 30:
out.append(f"2026/07/07 18:{i // 60 % 60:02d}:{i % 60:02d} [error] 812#0: SSL_do_handshake() "
f"failed (SSL: certificate verify failed: certificate has expired) "
f"while connecting to upstream auth-service:8443")
out.append(f'10.0.3.{r.randint(2, 250)} - - {ts} "GET /login HTTP/1.1" 502 552')
else:
out.append(f'10.0.3.{r.randint(2, 250)} - - {ts} "GET /{r.choice(["", "static/app.js", "api/health"])} '
f'HTTP/1.1" 200 {r.randint(200, 9000)}')
return "\n".join(out), r"expir|certificate"
def s_env_missing():
"""A crash-looping pod re-prints the same traceback 60 times; the KeyError
names the missing variable. Probe-failure noise is the red herring."""
out = []
for i in range(60):
out.append(f"2026-07-07T20:{i:02d}:01Z k8s: Readiness probe failed: connect: "
f"connection refused")
out.append(f"2026-07-07T20:{i:02d}:03Z k8s: Back-off restarting failed container "
f"app in pod shop-6d8f9/app")
out += ["Traceback (most recent call last):",
' File "/app/.venv/lib/python3.12/site-packages/myapp/config.py", line 44, '
"in load",
" dsn = os.environ['DATABASE_URL']",
' File "<frozen os>", line 685, in __getitem__',
"KeyError: 'DATABASE_URL'"]
return "\n".join(out), r"database_url|environment variable|env var"
def s_deadlock():
"""A 40-thread Java dump, hundreds of near-identical frames; one section
declares the deadlock."""
r, out = random.Random(18), []
for t in range(40):
out.append(f'"pool-1-thread-{t}" #{t + 20} prio=5 tid=0x{r.getrandbits(48):012x} '
f"waiting on condition")
for _ in range(12):
cls = r.choice(["QueueWorker", "BatchLoader", "HttpDispatch", "CacheSync"])
out.append(f"\tat com.shop.core.{cls}.run({cls}.java:{r.randint(40, 400)})")
out += ["Found one Java-level deadlock:",
'"OrderWriter" waiting to lock monitor 0x00007f2c (an InventoryLock),',
' which is held by "StockUpdater", which is waiting to lock 0x00007f2d,',
' which is held by "OrderWriter"']
for t in range(20):
out.append(f'"GC-thread-{t}" os_prio=31 tid=0x{r.getrandbits(48):012x} runnable')
return "\n".join(out), r"deadlock"
SCENARIOS = [("buried-oom", s_buried_oom), ("status-codes", s_status_codes),
("interleaved-disk", s_interleaved_disk), ("ansi-ci", s_ansi_ci),
("pip-conflict", s_pip_conflict), ("cert-expired", s_cert_expired),
("env-missing", s_env_missing), ("deadlock", s_deadlock)]
# ---------------------------------------------------------------- harness
def ask(prompt: str, cmd: str = MODEL_CMD) -> str:
r = subprocess.run(shlex.split(cmd), input=prompt, capture_output=True,
text=True, timeout=600)
if r.returncode != 0:
sys.exit(f"model command failed: {cmd}\n{r.stderr}")
return r.stdout
def main():
p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--dry", action="store_true",
help="build scenarios and report token counts; no model calls")
p.add_argument("--dump", metavar="DIR",
help="also write <name>.raw.txt / <name>.small.txt prompts to DIR")
a = p.parse_args()
from lessismore import _encoder
if _encoder() is None:
print("note: token counts are chars/4 estimates — pip install tiktoken "
"for o200k counts (grading is unaffected)", file=sys.stderr)
rows, ok_raw, ok_small = [], 0, 0
print(f"{'scenario':18} {'raw':>7} {'small':>7} {'ratio':>6} raw small")
for name, gen in SCENARIOS:
log, expect = gen()
small = compress(log, 2)
tr, ts = count_tokens(log), count_tokens(small)
if a.dump:
import pathlib
d = pathlib.Path(a.dump)
d.mkdir(exist_ok=True)
(d / f"{name}.raw.txt").write_text(log + "\n\n" + QUESTION)
(d / f"{name}.small.txt").write_text(small + "\n\n" + QUESTION)
if a.dry:
print(f"{name:18} {tr:>7,} {ts:>7,} {tr / ts:>5.1f}x")
continue
graded = []
for text in (log, small):
ans = ask(text + "\n\n" + QUESTION)
graded.append(bool(re.search(expect, ans, re.I)))
ok_raw += graded[0]
ok_small += graded[1]
mark = {True: "PASS", False: "FAIL"}
print(f"{name:18} {tr:>7,} {ts:>7,} {tr / ts:>5.1f}x {mark[graded[0]]} {mark[graded[1]]}")
rows.append((name, tr, ts, graded))
if not a.dry:
n = len(SCENARIOS)
print(f"\nroot cause found: raw {ok_raw}/{n}, compressed {ok_small}/{n} "
f"(model: {MODEL_CMD})")
if __name__ == "__main__":
main()

200
lessismore.egg-info/PKG-INFO Normal file
View File

@ -0,0 +1,200 @@
Metadata-Version: 2.4
Name: lessismore
Version: 0.1.0
Summary: Squeeze text before it hits an LLM — deterministic prompt compression, measured in real tokens
Author: John King (monsterrobotsoft)
License: MIT
Keywords: llm,prompt,compression,tokens,context
Requires-Python: >=3.8
Description-Content-Type: text/markdown
License-File: LICENSE
Provides-Extra: tokens
Requires-Dist: tiktoken; extra == "tokens"
Provides-Extra: ml
Requires-Dist: llmlingua; extra == "ml"
Dynamic: license-file
# lessismore 📉
**Turn 50,000 tokens of log spam into 4,500 tokens of pure signal — before it
ever hits your LLM.**
lessismore is a deterministic compression filter for the bulk text that
actually fills context windows: logs, CI output, tracebacks, JSON dumps,
captured tool output. Pure stdlib Python. Zero dependencies. Every claim below
was measured against the real o200k tokenizer — and every idea that failed the
measurement is documented at the bottom, so you know exactly what you're
getting.
```bash
tail -5000 app.log | python3 lessismore.py -l 2 | llm "why did this crash?"
```
**11x** on mixed error logs · **12x** on interleaved service logs ·
**8.9x** on ANSI CI logs · **98%** on captured pip/tqdm output ·
**~6 MB/s** single-core · **0** dependencies
## Why this exists
Three facts about LLM context, and the gap between them is this tool:
1. **The expensive part isn't your prompt.** Your typed question is ~30
tokens. The log dump you attach is 50,000. Compress the wall, not the
question.
2. **Machine output is mostly redundancy.** ANSI color codes, `\r` progress
redraws, timestamps, venv path spam, the same error line 400 times — noise
that looks small on a terminal screen but is real tokens in a capture.
3. **BPE tokenizers already compress English** (common words are 1 token —
you can't out-abbreviate them, we checked). What BPE *can't* see is
repetition across a file or terminal noise. That's the seam this tool
mines.
Three payoffs when you pipe through it:
- **Context capacity** — an hour of log history fits where two minutes did.
On a local model, that's the difference between full speed and crawling.
- **Prompt-cache longevity** — every pass is deterministic: same input, same
output, byte for byte. Follow-up questions re-hit the provider cache
instead of re-paying for the logs. An ML compressor in the loop would bust
the cache on every subtle variation; the regex passes never do.
- **Model attention** — LLMs lose things in the middle of walls of text. Feed
signal, not noise, and the first answer is the right one more often.
## Quickstart
```bash
pip install . # from a clone — installs the `lessismore` and `lm` commands
pip install '.[tokens]' # + tiktoken for exact token stats (else chars/4 estimate)
python3 test_lessismore.py # should print "ok"
```
```bash
lm dump.log -l 2 > small.log # file in, file out
docker logs myapp 2>&1 | lm -l 2 # pipe filter
lm big.txt -l 2 --budget 4000 # compress, then hard-cap at ~4k tokens
lm big.txt -l 2 --ml 0.5 # + LLMLingua last mile
lm --serve # paste-in demo page on localhost:7777
```
No install needed either — `python3 lessismore.py` works the same from a bare
clone; it's one stdlib-only file.
```python
from lessismore import compress, count_tokens, budget
small = budget(compress(big_log, level=2), 8000)
```
Token stats print to stderr, so pipes stay clean.
## The dial
Four levels, from byte-cautious to caveman. Pick by content, not by greed.
| Level | What it eats | Point it at | Profile |
|---|---|---|---|
| **1** | whitespace runs, consecutive duplicate lines | code, scripts, anything | structure-safe: indentation and content untouched (whitespace *inside string literals* still collapses) |
| **2** | + JSON minify, `\r` redraw collapse, ANSI strip, ISO timestamps, base64/hex blobs, UUIDs, venv paths, scattered-duplicate aliasing, similar-line collapse | logs, CI dumps, traces, tool output | **the sweet spot** — destroys machine noise, keeps every distinct fact |
| **3** | + filler-phrase stripping ("could you please", "just") | prose, chat history | fine for text, never for strict logic |
| **4** | + `two_sticks` caveman mode: drops articles/copulas/auxiliaries, never negations or modals | gist-only prose, transcripts | lossy on style, protective of meaning — "do **not** delete" keeps its *not* |
The crown jewel at level 2 is `alias_repeats`: ordinary dedupe only sees
*consecutive* repeats, so interleaved multi-service logs sail straight through
it. `alias_repeats` hunts scattered duplicates across the whole file and
dictionary-codes them (`@1 = ERROR [pool-3] psycopg2...` once in a legend,
2-token `@1` everywhere else). It's lossless — the legend keeps every line
verbatim — and on the interleaved benchmark it's the difference between
54,999 → 34,999 (dedupe alone, 1.6x) and 54,999 → 4,585 (**12x**).
`--budget N` is the backstop, not the compressor: after the passes run, it
keeps head and tail lines and drops the middle with a `[~N tokens omitted]`
marker. It's the only pass with *guaranteed* bounded output — use it when the
context limit is a hard wall.
## The receipts
All reproducible: `python3 bench.py` (o200k counts via tiktoken).
| Sample | Tokens | Why it wins |
|---|---|---|
| Mixed error log, 2000 lines | 86,914 → 7,895 (**11.0x**) | timestamp strip unmasks identical lines → dedupe; similar-line collapse catches the numbered stragglers |
| Interleaved 3-service log, zero consecutive repeats | 54,999 → 4,585 (**12.0x**) | `alias_repeats` — consecutive dedupe alone managed 1.6x on this input |
| ANSI-colored CI/docker build log | 21,198 → 2,387 (**8.9x**) | color codes make identical lines look different; strip them and the log collapses |
| Captured pip/tqdm output | 14,483 → 291 (**98%**) | every overwritten `\r` progress frame is invisible on screen but real tokens in a capture |
| pytest failure dump | 2,223 → 1,673 (25%) | one site-packages path prefix = 25 tokens → 7 |
| Pretty-printed JSON API response | 11,741 → 6,833 (42%) | minify (round-trip verified lossless) + UUID→8-hex squash |
| Chatty prose, level 4 | 427 → 264 (38%) | every dropped function word is a whole token; level 3 got 2% on the same text |
Throughput: measured **~6 MB/s single-core** at level 2 (28 MB log in 4.4s).
No model, no network — break-even input size is effectively zero.
## What we refused to build (measured so it stays dead)
The graveyard is a feature. Each of these looks clever on paper and loses
against a real tokenizer:
- **Shorthand codebooks** — `[fmt:md_tbl+hdr]` costs **8 tokens**; "Format
the output as a markdown table with headers." costs **10**. BPE already has
English baked in; bracket syntax shreds into off-distribution fragments.
- **Word→code recoding** ("database" → `qx`) — common words are already 1
token, random codes cost 2, plus ~4 tokens/entry of codebook tax. You
cannot beat a 200k-entry codebook from inside its own encoding. Zipping a
zip.
- **Known abbreviations** (db, fn, env, auth) — measured 1 → 1 tokens. Zero.
- **Personalized shorthand skills** — scanned 2,917 real typed prompts across
654 transcripts: 2,762 unique, and the repeats were already grunts ("yes",
"go"). A model-side decode skill costs ~1k tokens/turn to save ~10.
`grunts.py` keeps the half that works: mine your own history, emit
*client-side* slash-command stubs — expansion before the model sees it is
free.
- **Digit-masked dedupe** (0.0% — progress bars differ in glyphs, not
digits), **separator shortening** (a 78-char `----` is already 1 token),
**prefix hoisting** (one stray line kills it), **JSON→TSV / float
truncation / pointer squash** (real savings, worse trade).
## Battle-tested
An adversarial review agent was told to break it and confirmed 14 real
failure modes — negation-inverting filler stripping ("was **not just** the
db" → "was **not** the db"), URLs eaten as base64, crashes on empty and
non-UTF-8 stdin, distinct SHA-256s falsely merging as "repeated",
`two_sticks` eating "IT" and "US" as function words. Nine fixed with
regression tests, four documented below, one wontfix (adversarial in-band
marker collision). `python3 test_lessismore.py` — 30 asserts, no framework.
## Where it does nothing (on purpose)
Savings are proportional to **redundancy, not size**. This removes repetition
and machine noise; it cannot compress information, and doesn't pretend to.
| Content | Expect | Verdict |
|---|---|---|
| Repetitive machine output | 5–12x, up to 50x on pathological repeats | the reason this exists |
| Structured data (JSON, tracebacks) | 25–42% | worthwhile, lossless where it fires |
| Varied prose | ~2% (level 3) / 38% lossy (level 4) | gist mode only |
| Clean code, unique dense text | ~0% by design | that's what `--ml` or truncation is for |
`--ml RATE` bolts on Microsoft's LLMLingua-2 for perplexity pruning as a
last mile (`pip install llmlingua`) — runs *after* the deterministic passes so
you're not paying a classifier model to delete duplicate log lines. Only
worth it on multi-KB inputs, and it forfeits the cache-stability guarantee.
## Try it in a browser
`lm --serve` runs a paste-in demo page on `http://localhost:7777` — paste your
ugliest log, pick a level, watch the token count drop. Stdlib only, binds
localhost only.
## License
MIT — see [LICENSE](LICENSE).
## Known tradeoffs
- Levels 2+ assume line *order* matters but wall-clock timing doesn't. When
gaps and deltas are the signal (hang hunting), stay on level 1.
- Output is for LLM consumption, not round-tripping: markdown hard breaks
(trailing double-space) and diff context lines don't survive even level 1.
- At levels 3–4, dedupe counts describe the post-stripped text — five
differently-phrased "retry the job" lines can legitimately merge.
- In-band markers can collide with input that already contains them;
`alias_repeats` bails out if its own `@N` markers already appear as lines.

10
lessismore.egg-info/SOURCES.txt Normal file
View File

@ -0,0 +1,10 @@
LICENSE
README.md
lessismore.py
pyproject.toml
lessismore.egg-info/PKG-INFO
lessismore.egg-info/SOURCES.txt
lessismore.egg-info/dependency_links.txt
lessismore.egg-info/entry_points.txt
lessismore.egg-info/requires.txt
lessismore.egg-info/top_level.txt

1
lessismore.egg-info/dependency_links.txt Normal file
View File

@ -0,0 +1 @@

3
lessismore.egg-info/entry_points.txt Normal file
View File

@ -0,0 +1,3 @@
[console_scripts]
lessismore = lessismore:main
lm = lessismore:main

6
lessismore.egg-info/requires.txt Normal file
View File

@ -0,0 +1,6 @@
[ml]
llmlingua
[tokens]
tiktoken

1
lessismore.egg-info/top_level.txt Normal file
View File

@ -0,0 +1 @@
lessismore

281
lessismore.py
View File

@ -64,22 +64,8 @@ def dedupe_lines(text: str) -> str:
return "\n".join(out)
# no '/' in the class: URL paths are ≥64-char alnum+slash runs and they ARE the content.
# '-' and '_' admitted for base64url (real JWTs), guarded below so slugs survive.
_BLOB = re.compile(r"\b(?:[A-Za-z0-9+_-]{64,}={0,2}|[0-9a-fA-F]{48,})\b")
def _squash_blob(m: "re.Match") -> str:
s = m.group()
# a run with - or _ might be a kebab/snake/camelCase identifier, not a blob:
# demand blob-typical entropy (digits + both cases, no long alpha runs —
# 16+ consecutive letters ~never happens in base64, always in identifiers;
# the identifier case was caught by the LogDx-CI benchmark, not imagination)
if ("-" in s or "_" in s) and (not any(c.isdigit() for c in s)
or s.lower() == s or s.upper() == s
or re.search(r"[A-Za-z]{16,}", s)):
return s
return f"{s[:12]}[+{len(s) - 16} chars]{s[-4:]}"
# no '/' in the class: URL paths are ≥64-char alnum+slash runs and they ARE the content
_BLOB = re.compile(r"\b(?:[A-Za-z0-9+]{64,}={0,2}|[0-9a-fA-F]{48,})\b")
def squash_blobs(text: str) -> str:
@ -88,56 +74,26 @@ def squash_blobs(text: str) -> str:
The tail suffix keeps two different blobs from squashing to the same stub
and then being falsely merged as "repeated" by dedupe_lines.
"""
return _BLOB.sub(_squash_blob, text)
return _BLOB.sub(lambda m: f"{m.group()[:12]}[+{len(m.group()) - 16} chars]{m.group()[-4:]}", text)
_MON = r"(?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)"
# the last four formats were found by benchmarking on real LogHub logs
# (HDFS/BGL/Thunderbird/HealthApp); each is anchored enough not to eat data
_TIMESTAMP = re.compile(
r"\b\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}(?:[.,]\d+)?(?:Z|[+-]\d{2}:?\d{2})?\b ?" # ISO-8601
r"|\b" + _MON + r" {1,2}\d{1,2} \d{2}:\d{2}:\d{2}\b ?" # syslog
r"|\b\d{2}/" + _MON + r"/\d{4}:\d{2}:\d{2}:\d{2}(?: [+-]\d{4})?\b ?" # nginx CLF
r"|\b\d{9,10} \d{4}\.\d{2}\.\d{2}\b ?" # epoch+date pair (BGL/Thunderbird)
r"|\b\d{4}-\d{2}-\d{2}-\d{2}\.\d{2}\.\d{2}\.\d{6}\b ?" # BGL RAS event stamp
r"|(?m:^)\d{6} \d{6} (?=\d+ )" # HDFS 'yymmdd hhmmss pid'
r"|\b\d{8}-\d{2}:\d{2}:\d{2}:\d{3}\b ?" # HealthApp 'yyyymmdd-hh:mm:ss:ms'
r"\b\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}(?:[.,]\d+)?(?:Z|[+-]\d{2}:?\d{2})?\b ?"
)
def strip_timestamps(text: str) -> str:
"""Line order already encodes sequence; per-line timestamps are ~8 tokens each."""
# ponytail: epoch timestamps skipped on purpose — any 10-digit number matches
"""Line order already encodes sequence; per-line ISO timestamps are ~8 tokens each."""
# ponytail: ISO-8601 only; add syslog/other formats when a real log needs them
return _TIMESTAMP.sub("", text)
def minify_json(text: str) -> str:
"""Pretty JSON → minified, lossless when it fires, untouched when not.
Fires on the whole input AND on multi-line JSON embedded in other text
('response body:\\n{...}' — the common shape of captured tool output).
Single-line JSON (JSONL) is left alone: ', ' and ',' are 1 token either way;
only the newline+indent of pretty-printing costs real tokens.
"""
"""Whole-input pretty JSON → minified. Lossless when it fires, untouched when not."""
try:
return json.dumps(json.loads(text), separators=(",", ":"), ensure_ascii=False)
except ValueError:
pass
dec = json.JSONDecoder()
out, last = [], 0
for m in re.finditer(r"[{\[](?=[ \t]*\n)", text): # pretty JSON opens then breaks the line
if m.start() < last:
continue # inside a block we already minified
try:
obj, end = dec.raw_decode(text, m.start())
except ValueError:
continue
mini = json.dumps(obj, separators=(",", ":"), ensure_ascii=False)
if len(mini) < end - m.start():
out += [text[last:m.start()], mini]
last = end
out.append(text[last:])
return "".join(out)
return text
def collapse_cr(text: str) -> str:
@ -203,177 +159,26 @@ def two_sticks(text: str) -> str:
_SKEL = re.compile(r"[^A-Za-z]+")
_DIGITS = re.compile(r"\d+")
def _variant_summary(lines: list) -> str:
"""Digits are often THE signal (status codes, ports, counts) — summarize the
values a similar-line collapse would otherwise silently eat."""
runs = [_DIGITS.findall(l) for l in lines]
if not runs[0] or any(len(r) != len(runs[0]) for r in runs):
return ""
parts = []
for col in zip(*runs):
vals = sorted({int(v) for v in col})
if len(vals) == 1:
continue
parts.append("/".join(map(str, vals)) if len(vals) <= 4
else f"{vals[0]}–{vals[-1]}")
return "; values " + ", ".join(parts) if parts else ""
def _skel(l: str) -> str:
# hex runs masked first so goroutine addresses / template-ref hex values
# don't make otherwise-identical lines look distinct; a leading @tN ref
# name stays verbatim so different templates never merge as "similar"
m = _TREF.match(l)
head, tail = (l[:m.end()].strip() + " ", l[m.end():]) if m else ("", l)
return (head + _SKEL.sub(" ", _HEXRUN.sub(" ", tail)).strip()).strip()
def dedupe_similar(text: str) -> str:
"""Collapse runs of lines identical after masking hex runs and non-letters —
'same words, different numbers' (progress lines, per-item CI steps). Keeps
first and last so progression endpoints survive, and summarizes the varying
values in the marker so distinct facts (status codes, ports) aren't lost."""
"""Collapse runs of lines identical after masking non-letters — 'same words,
different numbers' (progress lines, per-item CI steps). Keeps first and last
so progression endpoints survive."""
lines = text.split("\n")
out, i = [], 0
while i < len(lines):
k, j = _skel(lines[i]), i
while j < len(lines) and _skel(lines[j]) == k:
k, j = _SKEL.sub(" ", lines[i]).strip(), i
while j < len(lines) and _SKEL.sub(" ", lines[j]).strip() == k:
j += 1
if j - i >= 4 and k:
out += [lines[i],
f"[{j - i - 2} similar lines omitted{_variant_summary(lines[i:j])}]",
lines[j - 1]]
out += [lines[i], f"[{j - i - 2} similar lines omitted]", lines[j - 1]]
else:
out.extend(lines[i:j])
i = j
return "\n".join(out)
_TREF = re.compile(r"@t\d+(?:\s|$)")
# a maximal hex-char run containing at least one digit: catches decimal runs,
# hex ids, and addresses, but never plain words ("added" is hex chars, no digit)
_HEXRUN = re.compile(r"[0-9a-fA-F]*[0-9][0-9a-fA-F]*")
_W = "\x00"
def _mine(lines, live, key_fn, val_fn, min_count):
"""One template-mining tier: group `live` line indexes by key_fn, keep
groups whose EXACT char accounting (original lines vs refs+legend) pays."""
groups = {}
for idx in live:
toks = lines[idx].split()
k, v = key_fn(toks), val_fn(toks)
if v:
g = groups.setdefault(k, [[], [], 0])
g[0].append(idx)
g[1].append(v)
g[2] += len(lines[idx])
worth = {} # insertion order = first appearance: numbering is deterministic
for k, (idxs, vs, orig_chars) in groups.items():
if len(vs) < min_count or len({len(v) for v in vs}) != 1:
continue
if sum(1 for t in k for c in t if c != _W) < 12: # junk guard
continue
# a value column that never varies (ssh2, a constant IP prefix)
# belongs in the template, not repeated in every ref line
const = [all(v[i] == vs[0][i] for v in vs) for i in range(len(vs[0]))]
it, tpl_toks = iter(range(len(vs[0]))), []
for t in k:
if _W not in t:
tpl_toks.append(t)
continue
parts = t.split(_W)
filled = parts[0]
for p in parts[1:]:
i = next(it)
filled += (vs[0][i] if const[i] else "<*>") + p
tpl_toks.append(filled)
tpl = " ".join(tpl_toks)
if "<*>" not in tpl: # all-constant = identical lines = alias_repeats territory
continue
ref_chars = sum(5 + sum(len(x) + 1 for x, c in zip(v, const) if not c)
for v in vs)
if orig_chars - ref_chars - (len(tpl) + 8) > 100: # exact payoff, not a guess
worth[k] = (tpl, const, idxs)
return worth
def _cheaper_whole(lines, tpl_c_idxs, key):
"""A mostly-varying token is cheaper carried whole (`10.251.91.84:52063`)
than as separate runs (`10 251 91 84 52063`). Re-cost the group with
whole-token values and switch encodings if that wins."""
tpl, const, idxs = tpl_c_idxs
wild_tok = [_W in t for t in key]
wvs = [tuple(t for t, w in zip(lines[i].split(), wild_tok) if w) for i in idxs]
wconst = [all(v[i] == wvs[0][i] for v in wvs) for i in range(len(wvs[0]))]
it = iter(range(len(wvs[0])))
wtpl = " ".join((wvs[0][i] if wconst[(i := next(it))] else "<*>") if w else t
for t, w in zip(key, wild_tok))
sub_cost = sum(sum(len(x) + 1 for x, c in zip(_subtok_vals(lines[i].split()), const)
if not c) for i in idxs) + len(tpl)
whole_cost = sum(sum(len(x) + 1 for x, c in zip(v, wconst) if not c)
for v in wvs) + len(wtpl)
if whole_cost < sub_cost and "<*>" in wtpl:
return (wtpl, wconst, idxs), "b"
return tpl_c_idxs, "a"
def _subtok_vals(toks):
return tuple(m for t in toks for m in _HEXRUN.findall(t))
def alias_templates(text: str, min_count: int = 4) -> str:
"""Two-tier Drain-lite template mining. Lossless modulo whitespace runs —
values keep their order, so a line reconstructs by filling <*> left to right.
Tier A masks hex/digit runs INSIDE tokens: `worker-3` and `10.2.3.44`
group as `worker-<*>` and `10.2.3.<*>`, and run-level constants (dates,
IP prefixes, the 2 in ssh2) inline into the template.
Tier B masks WHOLE digit-bearing tokens on whatever tier A left behind:
paths and ids that vary in letters (`.../hashtable.o` vs `.../sampler.o`)
still group, which is what crushes compile/build logs.
Both tiers were forced by public benchmarks (LogHub, LogDx-CI), not
imagination — each exists because a real corpus regressed without it.
"""
lines = text.split("\n")
if any(_TREF.match(l) for l in lines): # already templated, or real @tN content — bail
return text
live = range(len(lines))
worth_a = _mine(lines, live, lambda toks: tuple(_HEXRUN.sub(_W, t) for t in toks),
_subtok_vals, min_count)
taken = {i for _, _, idxs in worth_a.values() for i in idxs}
worth_b = _mine(lines, [i for i in live if i not in taken],
lambda toks: tuple(_W if any(c.isdigit() for c in t) else t
for t in toks),
lambda toks: tuple(t for t in toks if any(c.isdigit() for c in t)),
min_count)
if not (worth_a or worth_b):
return text
entries = [_cheaper_whole(lines, w, k) for k, w in worth_a.items()]
entries += [(w, "b") for w in worth_b.values()]
legend, ref_of = [], {}
for n, ((tpl, const, idxs), tier) in enumerate(entries, 1):
legend.append(f"@t{n} = {tpl}")
for i in idxs:
ref_of[i] = (f"@t{n}", const, tier)
out = []
for i, l in enumerate(lines):
r = ref_of.get(i)
if not r:
out.append(l)
continue
name, const, tier = r
toks = l.split()
v = _subtok_vals(toks) if tier == "a" else \
tuple(t for t in toks if any(c.isdigit() for c in t))
out.append(" ".join([name] + [x for x, c in zip(v, const) if not c]))
return "\n".join(["[templated lines, <*> = per-line values:]"] + legend + [""] + out)
_REF = re.compile(r"@\d+")
@ -402,7 +207,7 @@ def alias_repeats(text: str, min_count: int = 4, min_len: int = 30) -> str:
# normalized lines then match better in alias_repeats/dedupe_lines/dedupe_similar
_STRIP2 = [minify_json, collapse_cr, strip_ansi, strip_timestamps,
squash_blobs, squash_uuids, squash_pkgpaths]
_MERGE = [collapse_whitespace, alias_repeats, alias_templates, dedupe_lines, dedupe_similar]
_MERGE = [collapse_whitespace, alias_repeats, dedupe_lines, dedupe_similar]
LEVELS = {
1: [collapse_whitespace, dedupe_lines], # code-safe
2: _STRIP2 + _MERGE, # logs/dumps/tool output
@ -411,7 +216,7 @@ LEVELS = {
}
def compress(text: str, level: int = 2) -> str:
def compress(text: str, level: int = 1) -> str:
for f in LEVELS[level]:
text = f(text)
return text
@ -518,70 +323,27 @@ def compress_ml(text: str, rate: float = 0.5) -> str:
# ---------------------------------------------------------------- CLI
def run_command(cmd: str) -> "tuple[str, int]":
"""Run CMD in a shell, return (merged raw output, exit code) — exit code
survives, unlike piping through a filter."""
import subprocess
r = subprocess.run(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
return r.stdout.decode("utf-8", "replace"), r.returncode
def hook(level: int = 2, min_chars: int = 2000) -> None:
"""Claude Code PostToolUse hook mode: read the hook JSON on stdin, emit
updatedToolOutput JSON with the tool output compressed. Zero deps, no jq.
Small outputs pass through untouched (marker lines aren't worth the churn).
"""
try:
d = json.load(sys.stdin)
except ValueError:
return # not hook JSON — emit nothing, Claude Code keeps the original
out = d.get("tool_response") or d.get("tool_output") or {}
if isinstance(out, dict):
raw = out.get("stdout") or out.get("output") or ""
else:
raw = str(out)
if len(raw) < min_chars:
return
small = compress(raw, level)
if len(small) >= len(raw):
return
json.dump({"hookSpecificOutput": {"hookEventName": "PostToolUse",
"updatedToolOutput": {"stdout": small}}},
sys.stdout)
def main():
import argparse
p = argparse.ArgumentParser(prog="lessismore", description=__doc__.splitlines()[0])
p.add_argument("file", nargs="?", help="input file (default: stdin)")
p.add_argument("-l", "--level", type=int, default=2, choices=sorted(LEVELS),
help="1=code-safe 2=logs/dumps 3=prose 4=gist-only caveman (default 2)")
p.add_argument("-l", "--level", type=int, default=1, choices=sorted(LEVELS),
help="1=code-safe 2=logs/dumps 3=prose 4=gist-only caveman (default 1)")
p.add_argument("--ml", type=float, metavar="RATE",
help="also run LLMLingua-2 keeping RATE of tokens (needs: pip install llmlingua)")
p.add_argument("--budget", type=int, metavar="N",
help="hard cap output at ~N tokens: keep head+tail, drop the middle")
p.add_argument("--run", metavar="CMD",
help="run CMD in a shell, print its output compressed, exit with its status")
p.add_argument("--hook", action="store_true",
help="Claude Code PostToolUse hook mode (reads hook JSON on stdin)")
p.add_argument("--serve", nargs="?", const=7777, type=int, metavar="PORT",
help="serve a paste-in demo page on localhost (default port 7777)")
a = p.parse_args()
if a.serve:
return serve(a.serve)
if a.hook:
return hook(a.level)
if a.run:
raw, code = run_command(a.run)
else:
# newline="" / buffer.read(): keep \r intact for collapse_cr
raw = (open(a.file, encoding="utf-8", errors="replace", newline="").read() if a.file
else sys.stdin.buffer.read().decode("utf-8", "replace"))
code = 0
# newline="" / buffer.read(): keep \r intact for collapse_cr
raw = (open(a.file, encoding="utf-8", errors="replace", newline="").read() if a.file
else sys.stdin.buffer.read().decode("utf-8", "replace"))
out = compress(raw, a.level)
if a.ml:
try:
@ -594,7 +356,6 @@ def main():
before, after = count_tokens(raw), count_tokens(out)
print(f"lessismore: {before} → {after} tokens ({1 - after / max(before, 1):.0%} saved)",
file=sys.stderr)
sys.exit(code)
if __name__ == "__main__":

2
pyproject.toml
View File

@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "lessismore"
version = "0.4.0"
version = "0.1.0"
description = "Squeeze text before it hits an LLM — deterministic prompt compression, measured in real tokens"
readme = "README.md"
license = { text = "MIT" }

75
test_lessismore.py
View File

@ -65,34 +65,11 @@ def test():
# uuids shorten to 8-hex prefix
assert compress("id=550e8400-e29b-41d4-a716-446655440000 done\n", 2) == "id=550e8400… done\n"
# same-words-different-numbers: template mined, endpoints kept, ranges summarized
# same-words-different-numbers runs collapse, endpoints kept
sim = "\n".join(f"Downloading chunk {i} of 50 at {i * 3}kbps" for i in range(50))
out = compress(sim, 2)
# "of 50", "kbps", and "ssh2"-style run-level constants inline into the template
assert "@t1 = Downloading chunk <*> of 50 at <*>kbps" in out
assert "48 similar lines omitted; values 0–49, 0–147" in out
assert "@t1 0 0" in out and "@t1 49 147" in out
# scattered template repeats (the real-log shape): same message, different
# IPs/ports, interleaved with a second message — dictionary-coded as
# template + per-line values, and every varying value survives
ssh = "\n".join(f"Failed password for {['root', 'admin', 'guest'][i % 3]} "
f"from 10.2.3.{i} port {2200 + i} ssh2\n"
f"pam_unix(sshd:auth): authentication failure; rhost=10.2.3.{i}"
for i in range(24))
out = compress(ssh, 2)
# constant IP prefix and ssh2 inline; interleaved refs never merge as "similar"
assert "@t1 = Failed password for root from 10.2.3.<*> port <*> ssh2" in out
assert "\n@t1 0 2200\n" in out
for v in ("@t4 17 2217", "@t2 23"):
assert v in out, v
assert count_tokens(out) < count_tokens(ssh) / 1.5
# digits are often THE signal: distinct status codes survive the collapse
codes = "\n".join(f"ERROR upstream returned status {c} for /checkout"
for c in (500, 404, 503, 429, 500))
out = compress(codes, 2)
assert "404/429/500/503" in out
assert "[48 similar lines omitted]" in out
assert "chunk 0 of" in out and "chunk 49 of" in out
# whole-input pretty JSON minifies losslessly
import json
@ -101,32 +78,6 @@ def test():
assert json.loads(out) == json.loads(pretty)
assert len(out) < len(pretty) / 1.5
# pretty JSON EMBEDDED in other text minifies too — the captured-tool-output shape
emb = "response body:\n" + pretty + "\ndone in 0.31s"
out = compress(emb, 2)
assert out.startswith("response body:\n{") and out.endswith("done in 0.31s")
assert json.loads(out.split("\n")[1]) == json.loads(pretty)
# real-world JWTs are base64url (- and _) and get squashed...
jwt = "eyJhbGciOiJSUzI1NiIsImtpZCI6Il9abDVGdS0zOSJ9" * 3
assert "chars]" in squash_blobs("Authorization: Bearer " + jwt)
# ...but a long kebab-case slug is content, not a blob
slug = "-".join(["very", "long", "kebab", "case", "identifier"] * 4)
assert len(slug) >= 64 and squash_blobs(slug) == slug
# ...and so is a camelCase test name with digits (LogDx-CI regression)
ident = "codeFixMissingTypeAnnotationOnExports52-generics-oversimplifiedTypes"
assert squash_blobs(ident) == ident
# syslog and nginx CLF timestamps strip like ISO ones
assert strip_timestamps("Jul 7 10:00:00 host sshd[1]: boom") == "host sshd[1]: boom"
assert strip_timestamps('[07/Jul/2026:10:00:00 +0000] "GET /"') == '[] "GET /"'
# level 2 is idempotent: same input, same output, byte for byte — the cache claim
gnarly = ("2026-07-07T10:00:01Z ERROR conn refused db=orders retry\n" * 40
+ sim + "\n" + inter + "\x1b[32mok\x1b[0m\n")
once = compress(gnarly, 2)
assert compress(once, 2) == once
# budget: guaranteed cap, head and tail kept, middle dropped
text = "\n".join(f"line {i} of the log with some words" for i in range(200))
out = budget(text, 100)
@ -142,26 +93,6 @@ def test():
input=stdin, capture_output=True)
assert r.returncode == 0, r.stderr
# --run compresses the command's output and passes its exit code through
r = subprocess.run([sys.executable, "lessismore.py", "--run",
"printf 'a very long unique diagnostic line here\\n%.0s' 1 2 3 4 5; exit 7"],
capture_output=True, text=True)
assert r.returncode == 7
assert "repeated 4 more times" in r.stdout
# --hook: small outputs pass through untouched, big ones come back compressed
import json as _json
payload = _json.dumps({"tool_response": {"stdout": "tiny"}})
r = subprocess.run([sys.executable, "lessismore.py", "--hook"],
input=payload, capture_output=True, text=True)
assert r.returncode == 0 and r.stdout == ""
payload = _json.dumps({"tool_response": {"stdout":
"ERROR connection refused by upstream database\n" * 200}})
r = subprocess.run([sys.executable, "lessismore.py", "--hook"],
input=payload, capture_output=True, text=True)
hook_out = _json.loads(r.stdout)["hookSpecificOutput"]
assert "repeated 199 more times" in hook_out["updatedToolOutput"]["stdout"]
print("ok")