"""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.
# '-' 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:]}"
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(_squash_blob, 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'
)
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
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.
"""
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)
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"(? 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]+")
_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."""
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:
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]]
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+")
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, alias_templates, 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 = 2) -> 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 = ("lessismore"
""
"lessismore \U0001F4C9
"
"{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("{:,} → {:,} tokens ({}% saved)
"
"".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 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("--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
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)
sys.exit(code)
if __name__ == "__main__":
main()