mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
## 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>
221 lines
8 KiB
Python
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")
|