headroom/benchmarks/text_crusher_quality_eval.py
Zhenjia ZHOU 6c68ff4e9f
perf(compression): take large cold-start contexts off the synchronous kompress path (#1171) (#1298)
## Description

On a cold-start large context, kompress (ModernBERT ONNX) runs
**synchronously on the request thread** — ~200–300s for ~1M tokens. It
blows the 30s compression budget, leaks a non-preemptible worker, and
cascades (executor saturation → queue timeouts on healthy requests); on
timeout the request is forwarded **uncompressed** after eating 30s. This
adds four layered, **default-off, fail-open** mitigations so the request
path is never blocked on ML compression.

Closes #1171

## Type of Change

- [ ] Bug fix (non-breaking change that fixes an issue)
- [x] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [x] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

- **Phase 0 — size gate** (`HEADROOM_KOMPRESS_MAX_TOKENS`, default
50000): route oversized text away from ModernBERT (→ LogCompressor /
TextCrusher / passthrough) at the single `_try_ml_compressor` boundary.
- **Phase 1 — cooperative deadline**
(`HEADROOM_COMPRESSION_DEADLINE_MS`, default 20000): any kompress run
self-terminates at the next chunk boundary past the budget, keeping the
unprocessed tail verbatim.
- **Phase 2 — TextCrusher** (`HEADROOM_TEXT_CRUSHER`): a new **native
Rust** extractive prose compressor in
`crates/headroom-core/src/transforms/text_crusher/`, exposed via PyO3 as
`headroom._core.TextCrusher` with a thin Python wrapper. It **reuses the
shared `crate::relevance::BM25Scorer`** rather than reimplementing BM25,
and ships record/replay parity fixtures (mirroring the SmartCrusher
Rust-core + Python-shim pattern).
- **Phase 3 — off-path compression**
(`HEADROOM_BACKGROUND_COMPRESSION`): forward uncompressed immediately
and compress in a per-process background drain; a byte-identical cache
hit on a later turn means the request never blocks on ML.
- Benchmark (`benchmarks/text_crusher_quality_eval.py`), CHANGELOG
entry, and docstrings documenting the fail-open limits.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check`)
- [x] Type checking passes (`mypy`, new modules)
- [x] New tests added for new functionality
- [ ] Manual testing performed (Phase 0/1 gate-fire + deadline observed
on real traffic in earlier iterations; Phase 3 off-path is unit- +
byte-identity-tested, not yet live-validated)

### Test Output

```text
$ pytest tests/test_transforms/ tests/test_cache/ \
    tests/test_proxy/test_background_compression.py tests/test_proxy/test_phase3_byte_identity.py -q
501 passed, 37 skipped in 40.33s

$ cargo test -p headroom-core --lib text_crusher
test result: ok. 3 passed; 0 failed; 0 ignored; 0 measured; 834 filtered out

$ ruff check <changed files>
All checks passed!

$ mypy headroom/proxy/background_compression.py headroom/transforms/text_crusher.py
Success: no issues found in 2 source files
```

New coverage: size-gate incl. the strategy-dispatch funnel (KOMPRESS +
TEXT); partial-run deadline (chunk-0 compressed + chunk-1 verbatim
tail); BackgroundCompressor (dedup / queue-full / fail-open); Phase 3
byte-identity round-trip; TextCrusher unit + parity.

## Real Behavior Proof

- Environment: macOS, local dev — `uv` venv, Rust `_core` built via `uv
pip install -e .`.
- Exact command / steps: the `pytest` / `cargo test` / `ruff` / `mypy`
commands shown under Test Output; quality eval `python
benchmarks/text_crusher_quality_eval.py /tmp/squad_dev.json`.
- Observed result: 501 Python + 3 Rust tests pass; ruff + mypy clean on
changed/new modules. Quality eval: TextCrusher keeps ~94% of buried
SQuAD answers at 30% size vs ~36% truncate/random; self-contained speed
run ~333k words in ~76ms (one O(n) pass) — sub-second where ModernBERT
takes minutes (fast-vs-slow contrast, not a same-input run).
- Not tested: Phase 3 off-path on live traffic; multi-worker
(per-process by design — see Additional Notes).

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Checklist

- [x] My code follows the project's style guidelines
- [x] I have performed a self-review of my code
- [x] I have commented my code, particularly in hard-to-understand areas
- [x] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [x] I have updated the CHANGELOG.md if applicable

## Additional Notes

- **All four features are off by default and fail-open** — with the env
flags unset the paths are no-ops for realistic inputs; on any error the
request is forwarded (compressed if possible, else verbatim), never
dropped. A full background queue / duplicate key surfaces as
`deferred:dropped`.
- **Known limits (documented in `background_compression.py`):** Phase 3
is per-process, in-memory, and token-mode-only — these are
**lost-savings, never lost-correctness**, and consistent with the
project's existing per-process compression cache + sticky-session
multi-worker model. The startup multi-worker warning now names off-path
background compression.
- Phase 2 reuses the existing BM25 scorer; reuse did not improve
answer-retention over a Python prototype (query-awareness dominates) —
its value is the Rust speed + repo-conventional Rust-core/Python-shim
shape.

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-23 10:48:06 -05:00

221 lines
8 KiB
Python

#!/usr/bin/env python3
"""Quality eval for TextCrusher (Phase 2, #1171): does extractive compression
preserve the answer-bearing content? No LLM/API calls -- fully local.
Part A -- SQuAD answer-retention (the strong, labeled metric): bury a real QA
answer in a haystack of distractor paragraphs, compress to a target ratio, and
measure whether the gold answer SURVIVES. TextCrusher (query-aware) vs truncate
(keep-recent) vs random baselines. Mirrors kompress's published
must_keep_recall (0.977 on its own labeled set).
Part B -- real-transcript fidelity: compress large text blocks from a real
Claude Code transcript (ANONYMIZED), measuring ratio, speed, and salient-token
retention (identifiers/numbers/errors -- the must-keep info in coding contexts).
Only aggregate metrics are printed; raw content is never echoed.
Usage: python benchmarks/text_crusher_quality_eval.py [squad_dev.json] [transcript.jsonl]
"""
from __future__ import annotations
import glob
import json
import os
import random
import re
import sys
import time
from headroom.transforms.text_crusher import TextCrusher
_SEG = re.compile(r"(?<=[.!?])\s+|\n+")
_SALIENT = re.compile(
r"\b(?:error|exception|fail(?:ed|ure)?|warning|traceback|assert|todo|fixme)\b"
r"|\b[A-Z]{2,}\b|\b[A-Za-z_][A-Za-z0-9_]*\.[A-Za-z_][A-Za-z0-9_]*\b|\b\d+\b"
)
# --- anonymization (脱敏): scrub before any processing; never echo raw content ---
_REDACT = [
(re.compile(r"/Users/[^/\s]+"), "/Users/USER"),
(re.compile(r"\b[\w.+-]+@[\w-]+\.[\w.-]+\b"), "EMAIL"),
(re.compile(r"\b(?:sk|pk|ghp|gho|xox[baprs])-[A-Za-z0-9_-]{10,}\b"), "TOKEN"),
(re.compile(r"\bBearer\s+[A-Za-z0-9._-]{10,}"), "Bearer TOKEN"),
(re.compile(r"\b[A-Fa-f0-9]{40,}\b"), "HEX"),
]
def anon(t: str) -> str:
for rx, rep in _REDACT:
t = rx.sub(rep, t)
return t
def norm(s: str) -> str:
return re.sub(r"\s+", " ", s.lower()).strip()
def _segs(text: str) -> list[str]:
return [s for s in _SEG.split(text) if s.strip()]
def truncate_keep_last(text: str, ratio: float) -> str:
segs = _segs(text)
budget = int(sum(len(s) for s in segs) * ratio)
kept: list[str] = []
c = 0
for s in reversed(segs):
if c >= budget:
break
kept.append(s)
c += len(s)
return "\n".join(reversed(kept))
def random_keep(text: str, ratio: float, seed: int) -> str:
segs = _segs(text)
idx = list(range(len(segs)))
random.Random(seed).shuffle(idx)
budget = int(sum(len(s) for s in segs) * ratio)
kept: set[int] = set()
c = 0
for i in idx:
if c >= budget:
break
kept.add(i)
c += len(segs[i])
return "\n".join(segs[i] for i in sorted(kept))
def eval_squad(path: str, n: int = 200, n_distract: int = 40, ratio: float = 0.3, seed: int = 0):
data = json.load(open(path))
paras = [(p["context"], p["qas"]) for a in data["data"] for p in a["paragraphs"]]
all_ctx = [c for c, _ in paras]
examples = [
(ctx, qas[0]["question"], qas[0]["answers"][0]["text"])
for ctx, qas in paras
if qas and qas[0]["answers"]
]
rnd = random.Random(seed)
rnd.shuffle(examples)
examples = examples[:n]
tc = TextCrusher()
hit = {"text_crusher": 0, "truncate": 0, "random": 0}
tc_ratios: list[float] = []
for gold_ctx, q, ans in examples:
docs = rnd.sample(all_ctx, n_distract) + [gold_ctx]
rnd.shuffle(docs)
haystack = "\n\n".join(docs)
a = norm(ans)
out_tc = tc.compress(haystack, context=q, target_ratio=ratio).compressed
hit["text_crusher"] += a in norm(out_tc)
hit["truncate"] += a in norm(truncate_keep_last(haystack, ratio))
hit["random"] += a in norm(random_keep(haystack, ratio, seed))
tc_ratios.append(len(out_tc) / max(1, len(haystack)))
nn = len(examples)
print(
f"\n=== Part A: SQuAD answer-retention (n={nn}, distractors={n_distract}, target_ratio={ratio}) ==="
)
print(
f" TextCrusher (query-aware): {hit['text_crusher'] / nn:6.1%} answer survives compression"
)
print(f" Truncate (keep recent): {hit['truncate'] / nn:6.1%}")
print(f" Random keep: {hit['random'] / nn:6.1%}")
print(
f" TextCrusher mean char-ratio: {sum(tc_ratios) / nn:.2f} (kept ~{sum(tc_ratios) / nn:.0%} of bytes)"
)
print(" reference: kompress published must_keep_recall = 0.977 (its own labeled set)")
def _block_texts(jsonl_path: str, min_words: int, limit: int) -> list[str]:
out: list[str] = []
with open(jsonl_path) as fh:
for line in fh:
try:
o = json.loads(line)
except json.JSONDecodeError:
continue
m = o.get("message") or {}
c = m.get("content")
parts = (
[c]
if isinstance(c, str)
else [
p["text"] for p in c if isinstance(p, dict) and isinstance(p.get("text"), str)
]
if isinstance(c, list)
else []
)
for t in parts:
if len(t.split()) >= min_words:
out.append(anon(t))
if len(out) >= limit:
break
return out[:limit]
def eval_transcript(jsonl_path: str, ratio: float = 0.4, min_words: int = 1500, limit: int = 40):
blocks = _block_texts(jsonl_path, min_words, limit)
if not blocks:
print(
f"\n=== Part B: no text blocks >= {min_words} words in {os.path.basename(jsonl_path)} ==="
)
return
tc = TextCrusher()
ratios: list[float] = []
times: list[float] = []
retentions: list[float] = []
for b in blocks:
sal_before = set(_SALIENT.findall(b))
t0 = time.perf_counter()
out = tc.compress(b, target_ratio=ratio).compressed
times.append((time.perf_counter() - t0) * 1000)
sal_after = set(_SALIENT.findall(out))
retentions.append(len(sal_before & sal_after) / max(1, len(sal_before)))
ratios.append(len(out.split()) / max(1, len(b.split())))
n = len(blocks)
print(
f"\n=== Part B: real transcript fidelity (n={n} large blocks, anonymized, target_ratio={ratio}) ==="
)
print(f" mean token-ratio kept: {sum(ratios) / n:.2f}")
print(f" mean speed: {sum(times) / n:.1f} ms/block")
print(
f" salient-token retention: {sum(retentions) / n:6.1%} (identifiers/numbers/errors kept)"
)
print(
f" -> keeps salient info at {sum(retentions) / n:.0%} while dropping to {sum(ratios) / n:.0%} of tokens"
)
def eval_speed(scale_words: int = 250_000):
# Reproducible throughput on a large synthetic prose block (no external data).
text = " ".join(
f"Sentence {i} discusses subsystem {i} and its failure mode {i % 7} in detail."
for i in range(scale_words // 9)
)
nwords = len(text.split())
tc = TextCrusher()
t0 = time.perf_counter()
out = tc.compress(text, target_ratio=0.3)
ms = (time.perf_counter() - t0) * 1000
print(f"\n=== Part C: speed (synthetic, {nwords:,} words, fully reproducible) ===")
print(f" TextCrusher compress: {ms:.0f} ms ({nwords / max(ms / 1000, 1e-6):,.0f} words/sec)")
print(f" kept ratio: {out.compressed_tokens / max(1, out.original_tokens):.2f}")
print(" reference: kompress (ModernBERT ONNX) ~272s for ~1M tokens (measured, query-blind)")
print(" -> fast-vs-slow CONTRAST, not a same-input side-by-side run")
if __name__ == "__main__":
eval_speed()
squad = sys.argv[1] if len(sys.argv) > 1 else "/tmp/squad_dev.json"
tx = sys.argv[2] if len(sys.argv) > 2 else None
if os.path.exists(squad):
eval_squad(squad)
else:
print(f"SQuAD not found at {squad}; skipping Part A")
if tx is None:
found = glob.glob(os.path.expanduser("~/.claude/projects/*headroom*/*.jsonl"))
tx = max(found, key=os.path.getsize) if found else None
if tx and os.path.exists(tx):
eval_transcript(tx)
else:
print("no transcript jsonl found; skipping Part B")