"""Command-line interface for aave-compress. Usage:: aave-compress "some text" --level medium --model claude-sonnet-5 aave-compress "some text" --level full --strip-filler --tiktoken-model gpt-4o --gate Prints the original and compressed text, the markers that fired, and — only when a real tokenizer count is available via :mod:`aave_compress.tokenizers` — the measured token savings. There is deliberately NO estimated-savings fallback based on characters or words: unverified compression claims are exactly the failure mode this project exists to avoid. ``--gate`` needs a working tokenizer (from ``--model`` or the approximate ``--tiktoken-model`` fallback). If neither works, the run proceeds UNGATED with a warning rather than crashing. """ from __future__ import annotations import argparse import sys from collections.abc import Callable from .engine import LEVELS, compress from .tokenizers import count_tokens, count_tokens_tiktoken def _resolve_counter( args: argparse.Namespace, ) -> tuple[Callable[[str], int] | None, str]: """Probe tokenizer availability once, up front. Returns ``(counter_fn, label)``. When no tokenizer works, ``counter_fn`` is None and ``label`` carries the reason. """ probe = "tokenizer availability probe" try: probed = count_tokens(probe, model=args.model) return ( lambda text: count_tokens(text, model=args.model).tokens, f"{args.model}, method={probed.method}", ) except Exception as exc: # noqa: BLE001 - probe: any failure means "fall back", never crash # Covers missing-key RuntimeError/ValueError *and* API-side failures # (auth, zero credit balance, rate limit, network) raised by the # anthropic SDK, which are not RuntimeError/ValueError subclasses. reason = str(exc) if args.tiktoken_model: try: count_tokens_tiktoken(probe, model=args.tiktoken_model) except Exception as exc: # noqa: BLE001 - same probe-never-crashes rule return None, f"{reason} (tiktoken fallback also failed: {exc})" return ( lambda text: count_tokens_tiktoken( text, model=args.tiktoken_model ).tokens, f"{args.tiktoken_model}, method=tiktoken_approx — GPT-family " "approximation, NOT valid for any Claude model", ) return None, reason def main(argv: list[str] | None = None) -> None: parser = argparse.ArgumentParser( prog="aave-compress", description=( "Compress LLM prompt text using documented AAVE grammar patterns " "and measure real token savings." ), ) parser.add_argument("text", help="Text to compress") parser.add_argument( "--level", choices=sorted(LEVELS), default="medium", help="Compression level (default: medium)", ) parser.add_argument( "--model", default="claude-sonnet-5", help=( "Model to count tokens for (default: claude-sonnet-5). claude-* " "models use the Anthropic count_tokens API (requires " "ANTHROPIC_API_KEY); gpt-* models use tiktoken." ), ) parser.add_argument( "--tiktoken-model", default=None, metavar="GPT_MODEL", help=( "APPROXIMATE fallback: count tokens with tiktoken for this " "GPT-family model (e.g. gpt-4o) if the primary counter is " "unavailable. NEVER valid for Claude models." ), ) parser.add_argument( "--strip-filler", action="store_true", help=( "Run the generic filler/hedge/politeness stripper " "(filters.strip_fillers) as a first pass before the AAVE markers." ), ) parser.add_argument( "--gate", action="store_true", help=( "Keep each transform step only if it strictly reduces the token " "count. Needs a working tokenizer (--model or --tiktoken-model); " "without one the run proceeds ungated." ), ) args = parser.parse_args(argv) if args.tiktoken_model and args.tiktoken_model.lower().startswith("claude"): parser.error( "--tiktoken-model must be a GPT-family model: tiktoken counts are " "never valid for Claude models (they undercount by 15-20%+). " "Set ANTHROPIC_API_KEY and use --model instead." ) counter, counter_label = _resolve_counter(args) gate = args.gate if gate and counter is None: print( "Gating needs a working tokenizer, and none is available: " f"{counter_label}\nProceeding UNGATED.", file=sys.stderr, ) gate = False result = compress( args.text, level=args.level, strip_filler=args.strip_filler, tokenizer=counter, gate=gate, ) print(f"Level: {result.level}") if args.strip_filler: state = "fillers removed" if result.filler_stripped else "nothing to remove" print(f"Filler strip: on ({state})") else: print("Filler strip: off") if result.gated: print( "Gating: on — each step kept only if it strictly reduced the " f"token count ({counter_label})" ) elif args.gate: print("Gating: requested but ran UNGATED (no working tokenizer)") else: print("Gating: off") print() print("Original:") print(f" {result.original}") print() print("Compressed:") print(f" {result.compressed}") print() if result.markers_applied: print("Markers applied: " + ", ".join(result.markers_applied)) else: print("Markers applied: (none)") print() if counter is None: print( "Token counts unavailable: set ANTHROPIC_API_KEY to see real " "savings — do not trust word-count-based estimates as a " f"substitute. ({counter_label})", file=sys.stderr, ) return try: original_tokens = counter(result.original) compressed_tokens = counter(result.compressed) except Exception as exc: print( f"Token counting failed mid-run ({type(exc).__name__}: {exc}); " "no counts reported.", file=sys.stderr, ) return ratio = ( compressed_tokens / original_tokens if original_tokens else float("nan") ) print(f"Token counts ({counter_label}):") print(f" Original: {original_tokens}") print(f" Compressed: {compressed_tokens}") print(f" Ratio (compressed/original): {ratio:.3f}") print() print( "NOTE: this ratio is for THIS ONE INPUT ONLY — a single example, NOT a " "benchmark. Do not quote it as a general savings figure. For the " "corpus-level number (currently 5.0% mean at level=full with " "--strip-filler, n=20 non-cherry-picked prompts, GPT-4o " "tiktoken-approximate — not yet Claude-verified) run " "`python eval/run_eval.py`; see docs/methodology.md." ) if __name__ == "__main__": main()