#!/usr/bin/env python3 """ Agent Cost Crisis Benchmark - The Compelling Story This benchmark demonstrates WHY Headroom matters by showing: 1. THE PROBLEM: Context explosion in real-world agent workloads - Tokens grow exponentially with conversation length - Tool outputs dominate context (often 70%+ of tokens) - Dynamic content breaks cache efficiency 2. THE SOLUTION: Headroom's impact on real workloads - Token reduction from SmartCrusher (50-80% on tool outputs) - Cache alignment improvement (10x+ potential savings) - Context windowing (stay within limits without losing info) 3. THE PROOF: Quality preservation - Critical information retained (errors, anomalies, relevant items) - Agent task completion unaffected - Information retrieval accuracy maintained Usage: python benchmarks/agent_cost_benchmark.py python benchmarks/agent_cost_benchmark.py --format markdown > BENCHMARK.md python benchmarks/agent_cost_benchmark.py --scenario coding-agent """ from __future__ import annotations import argparse import json import statistics import time from dataclasses import dataclass, field from typing import Any # Benchmark scenario imports from benchmarks.scenarios.conversations import ( generate_agentic_conversation, generate_rag_conversation, ) from benchmarks.scenarios.tool_outputs import ( generate_log_entries, generate_search_results, ) # Headroom imports from headroom.transforms.smart_crusher import SmartCrusherConfig, smart_crush_tool_output # ============================================================================= # PRICING DATA (as of 2025) # ============================================================================= PRICING = { # Anthropic Claude 3.5 Sonnet "claude-3.5-sonnet": { "input": 3.00 / 1_000_000, # $3 per 1M tokens "output": 15.00 / 1_000_000, # $15 per 1M tokens "cached_input": 0.30 / 1_000_000, # 90% discount on cache hit "cache_write": 3.75 / 1_000_000, # 25% premium to write cache }, # OpenAI GPT-4o "gpt-4o": { "input": 2.50 / 1_000_000, "output": 10.00 / 1_000_000, "cached_input": 1.25 / 1_000_000, # 50% discount }, # Google Gemini 1.5 Pro "gemini-1.5-pro": { "input": 1.25 / 1_000_000, "output": 5.00 / 1_000_000, "cached_input": 0.3125 / 1_000_000, # 75% discount }, } # Approximate tokens per character (GPT-4 tokenizer average) CHARS_PER_TOKEN = 4 @dataclass class CostAnalysis: """Cost analysis for a workload.""" tokens_input: int = 0 tokens_output: int = 0 tokens_cached: int = 0 cost_baseline: float = 0.0 cost_optimized: float = 0.0 cost_with_cache: float = 0.0 savings_from_compression: float = 0.0 savings_from_caching: float = 0.0 total_savings_percent: float = 0.0 @dataclass class BenchmarkResult: """Result from a single benchmark scenario.""" name: str description: str # Token metrics tokens_original: int = 0 tokens_optimized: int = 0 compression_ratio: float = 0.0 # Cache metrics cache_hit_rate_baseline: float = 0.0 cache_hit_rate_optimized: float = 0.0 # Quality metrics critical_items_retained: int = 0 critical_items_total: int = 0 retention_rate: float = 0.0 # Cost analysis cost_analysis: CostAnalysis = field(default_factory=CostAnalysis) # Performance optimization_latency_ms: float = 0.0 # Details details: dict[str, Any] = field(default_factory=dict) # ============================================================================= # SCENARIO 1: Coding Agent Context Explosion # ============================================================================= def benchmark_coding_agent_explosion() -> BenchmarkResult: """ Simulate a Claude Code / Cursor style coding agent session. Shows how context explodes as the agent: - Searches codebase (100s of file snippets) - Reads documentation (large text blocks) - Makes tool calls (grep, find, read) - Accumulates conversation history """ result = BenchmarkResult( name="Coding Agent Context Explosion", description="50-turn coding session with file search, grep, and documentation lookups", ) # Generate realistic coding agent conversation messages = generate_agentic_conversation( turns=50, tool_calls_per_turn=2, items_per_tool_response=100, # 100 search results per tool call ) # Calculate original tokens original_content = json.dumps(messages) result.tokens_original = len(original_content) // CHARS_PER_TOKEN # Apply Headroom transforms using convenience function config = SmartCrusherConfig(max_items_after_crush=20) start = time.perf_counter() optimized_messages = [] critical_retained = 0 critical_total = 0 for msg in messages: if msg.get("role") == "tool": # Parse tool content as JSON array try: original_content = msg.get("content", "[]") content = json.loads(original_content) if isinstance(content, list) and len(content) > 10: # Count critical items (errors, high-relevance) for item in content: if isinstance(item, dict): if item.get("error") or item.get("status") == "failed": critical_total += 1 if item.get("is_needle"): critical_total += 1 # Compress with SmartCrusher convenience function compressed_str, was_modified, _ = smart_crush_tool_output( original_content, config ) if was_modified: compressed = json.loads(compressed_str) # Count retained critical items for item in compressed: if isinstance(item, dict): if item.get("error") or item.get("status") == "failed": critical_retained += 1 if item.get("is_needle"): critical_retained += 1 msg = {**msg, "content": compressed_str} except (json.JSONDecodeError, TypeError): pass optimized_messages.append(msg) result.optimization_latency_ms = (time.perf_counter() - start) * 1000 # Calculate optimized tokens optimized_content = json.dumps(optimized_messages) result.tokens_optimized = len(optimized_content) // CHARS_PER_TOKEN # Calculate metrics result.compression_ratio = 1 - (result.tokens_optimized / result.tokens_original) result.critical_items_total = critical_total result.critical_items_retained = critical_retained result.retention_rate = critical_retained / critical_total if critical_total > 0 else 1.0 # Cost analysis (using Claude 3.5 Sonnet pricing) pricing = PRICING["claude-3.5-sonnet"] result.cost_analysis = CostAnalysis( tokens_input=result.tokens_original, cost_baseline=result.tokens_original * pricing["input"], cost_optimized=result.tokens_optimized * pricing["input"], savings_from_compression=(result.tokens_original - result.tokens_optimized) * pricing["input"], ) result.cost_analysis.total_savings_percent = result.compression_ratio * 100 result.details = { "turns": 50, "tool_calls": 100, "items_per_response": 100, "items_after_compression": 20, } return result # ============================================================================= # SCENARIO 2: Cache Alignment Impact # ============================================================================= def benchmark_cache_alignment() -> BenchmarkResult: """ Show how dynamic content breaks caching and how CacheAligner fixes it. Simulates 100 requests with same base prompt but different dates. Without alignment: 0% cache hits With alignment: 90%+ cache hits """ from headroom.cache import DetectorConfig, DynamicContentDetector result = BenchmarkResult( name="Cache Alignment Impact", description="100 requests with dynamic dates - cache hit improvement", ) # Base system prompt with dynamic date base_prompt = """You are Claude, an AI assistant by Anthropic. Today is {date}. Current time: {time}. Session ID: {session_id} Request ID: {request_id} You are a helpful coding assistant. Follow these guidelines: 1. Write clean, readable code 2. Add appropriate comments 3. Handle errors gracefully 4. Follow best practices Be concise and helpful.""" import datetime import uuid # Use DynamicContentDetector to extract static content detector = DynamicContentDetector(DetectorConfig(tiers=["regex"])) # Simulate 100 requests over a day prompts_original = [] prompts_aligned = [] base_date = datetime.datetime(2025, 1, 15, 9, 0, 0) for i in range(100): # Each request has different timestamp request_time = base_date + datetime.timedelta(minutes=i * 5) prompt = base_prompt.format( date=request_time.strftime("%A, %B %d, %Y"), time=request_time.strftime("%I:%M %p"), session_id=f"sess_{uuid.uuid4().hex[:24]}", request_id=f"req_{uuid.uuid4().hex[:24]}", ) prompts_original.append(prompt) # Extract static content for cache alignment detection_result = detector.detect(prompt) prompts_aligned.append(detection_result.static_content) # Calculate cache hits # Baseline: all prompts are different (dynamic dates) unique_original = len(set(prompts_original)) cache_hits_baseline = 100 - unique_original # Aligned: static prefixes should be identical unique_aligned = len(set(prompts_aligned)) cache_hits_aligned = 100 - unique_aligned result.cache_hit_rate_baseline = cache_hits_baseline / 100 result.cache_hit_rate_optimized = cache_hits_aligned / 100 # Token calculation result.tokens_original = sum(len(p) // CHARS_PER_TOKEN for p in prompts_original) # Cost analysis with caching pricing = PRICING["claude-3.5-sonnet"] tokens_per_request = len(prompts_original[0]) // CHARS_PER_TOKEN # Baseline: pay full price every time (no cache hits) cost_baseline = 100 * tokens_per_request * pricing["input"] # Optimized: first request is cache write, rest are cache hits first_request_cost = tokens_per_request * pricing["cache_write"] cached_requests_cost = 99 * tokens_per_request * pricing["cached_input"] cost_optimized = first_request_cost + cached_requests_cost result.cost_analysis = CostAnalysis( tokens_input=result.tokens_original, cost_baseline=cost_baseline, cost_with_cache=cost_optimized, savings_from_caching=cost_baseline - cost_optimized, total_savings_percent=((cost_baseline - cost_optimized) / cost_baseline) * 100, ) result.details = { "total_requests": 100, "unique_prompts_baseline": unique_original, "unique_prompts_aligned": unique_aligned, "cache_improvement_factor": f"{(cache_hits_aligned - cache_hits_baseline)}x", } return result # ============================================================================= # SCENARIO 3: RAG Context Scaling # ============================================================================= def benchmark_rag_scaling() -> BenchmarkResult: """ Show how RAG context grows and how Headroom manages it. Simulates large RAG context with multiple queries. """ result = BenchmarkResult( name="RAG Context Scaling", description="Large RAG context (~50K tokens) with compression" ) # Generate RAG conversation with ~50K tokens of context messages = generate_rag_conversation( context_tokens=50000, num_queries=10, ) original_content = json.dumps(messages) result.tokens_original = len(original_content) // CHARS_PER_TOKEN # Apply transforms - compress tool outputs in messages config = SmartCrusherConfig(max_items_after_crush=10) start = time.perf_counter() # Compress tool outputs in messages optimized_messages = [] for msg in messages: if msg.get("role") == "tool": try: original_content_msg = msg.get("content", "[]") compressed_str, was_modified, _ = smart_crush_tool_output( original_content_msg, config ) if was_modified: msg = {**msg, "content": compressed_str} except Exception: pass optimized_messages.append(msg) result.optimization_latency_ms = (time.perf_counter() - start) * 1000 optimized_content = json.dumps(optimized_messages) result.tokens_optimized = len(optimized_content) // CHARS_PER_TOKEN result.compression_ratio = 1 - (result.tokens_optimized / result.tokens_original) # Cost analysis pricing = PRICING["claude-3.5-sonnet"] result.cost_analysis = CostAnalysis( tokens_input=result.tokens_original, cost_baseline=result.tokens_original * pricing["input"], cost_optimized=result.tokens_optimized * pricing["input"], savings_from_compression=(result.tokens_original - result.tokens_optimized) * pricing["input"], total_savings_percent=result.compression_ratio * 100, ) result.details = { "context_tokens": 50000, "num_queries": 10, } return result # ============================================================================= # SCENARIO 4: Long-Running Agent Session # ============================================================================= def benchmark_conversation_scaling() -> list[BenchmarkResult]: """ Show how costs scale with conversation length. Generates conversations of increasing length (10, 25, 50, 100, 200 turns) and shows the scaling curve with and without Headroom. """ results = [] turn_counts = [10, 25, 50, 100, 200] for turns in turn_counts: result = BenchmarkResult( name=f"Conversation Scaling ({turns} turns)", description=f"{turns}-turn agent conversation with tool calls", ) messages = generate_agentic_conversation( turns=turns, tool_calls_per_turn=1, items_per_tool_response=50, ) original_content = json.dumps(messages) result.tokens_original = len(original_content) // CHARS_PER_TOKEN # Apply full optimization pipeline config = SmartCrusherConfig(max_items_after_crush=15) start = time.perf_counter() optimized = [] for msg in messages: if msg.get("role") == "tool": try: original_content = msg.get("content", "[]") content = json.loads(original_content) if isinstance(content, list) and len(content) > 15: compressed_str, was_modified, _ = smart_crush_tool_output( original_content, config ) if was_modified: msg = {**msg, "content": compressed_str} except (json.JSONDecodeError, TypeError): pass optimized.append(msg) result.optimization_latency_ms = (time.perf_counter() - start) * 1000 optimized_content = json.dumps(optimized) result.tokens_optimized = len(optimized_content) // CHARS_PER_TOKEN result.compression_ratio = 1 - (result.tokens_optimized / result.tokens_original) pricing = PRICING["claude-3.5-sonnet"] result.cost_analysis = CostAnalysis( tokens_input=result.tokens_original, cost_baseline=result.tokens_original * pricing["input"], cost_optimized=result.tokens_optimized * pricing["input"], total_savings_percent=result.compression_ratio * 100, ) result.details = {"turns": turns} results.append(result) return results # ============================================================================= # SCENARIO 5: Quality Preservation Test # ============================================================================= def benchmark_quality_preservation() -> BenchmarkResult: """ Prove that compression doesn't lose critical information. Generates data with known "needles" (errors, anomalies, high-relevance items) and verifies they survive compression. """ result = BenchmarkResult( name="Quality Preservation", description="Verify critical items (errors, anomalies) survive compression", ) # Generate test data with known needles search_results = generate_search_results( n=1000, include_uuid_needles=10, include_errors=20, ) log_entries = generate_log_entries( n=1000, include_errors=30, include_critical=5, ) # Count needles before compression needles_before = 0 errors_before = 0 for item in search_results: if item.get("is_needle"): needles_before += 1 if item.get("error"): errors_before += 1 for entry in log_entries: if entry.get("level") in ("ERROR", "CRITICAL"): errors_before += 1 # Compress using SmartCrusher convenience function config = SmartCrusherConfig(max_items_after_crush=50) search_str = json.dumps(search_results) logs_str = json.dumps(log_entries) compressed_search_str, _, _ = smart_crush_tool_output(search_str, config) compressed_logs_str, _, _ = smart_crush_tool_output(logs_str, config) compressed_search = json.loads(compressed_search_str) compressed_logs = json.loads(compressed_logs_str) # Count needles after compression needles_after = 0 errors_after = 0 for item in compressed_search: if item.get("is_needle"): needles_after += 1 if item.get("error"): errors_after += 1 for entry in compressed_logs: if entry.get("level") in ("ERROR", "CRITICAL"): errors_after += 1 result.critical_items_total = needles_before + errors_before result.critical_items_retained = needles_after + errors_after result.retention_rate = result.critical_items_retained / result.critical_items_total result.tokens_original = ( len(json.dumps(search_results)) + len(json.dumps(log_entries)) ) // CHARS_PER_TOKEN result.tokens_optimized = ( len(json.dumps(compressed_search)) + len(json.dumps(compressed_logs)) ) // CHARS_PER_TOKEN result.compression_ratio = 1 - (result.tokens_optimized / result.tokens_original) result.details = { "search_results_original": 1000, "search_results_compressed": len(compressed_search), "log_entries_original": 1000, "log_entries_compressed": len(compressed_logs), "needles_original": needles_before, "needles_retained": needles_after, "errors_original": errors_before, "errors_retained": errors_after, } return result # ============================================================================= # REPORT GENERATION # ============================================================================= def generate_report(results: list[BenchmarkResult], format: str = "terminal") -> str: """Generate benchmark report in specified format.""" if format == "markdown": return _generate_markdown_report(results) else: return _generate_terminal_report(results) def _generate_terminal_report(results: list[BenchmarkResult]) -> str: """Generate colorful terminal report.""" lines = [] lines.append("") lines.append("=" * 80) lines.append(" HEADROOM AGENT COST BENCHMARK") lines.append(" The Context Optimization Layer for LLM Applications") lines.append("=" * 80) total_savings = 0.0 total_baseline = 0.0 for result in results: lines.append("") lines.append(f"{'─' * 80}") lines.append(f" {result.name}") lines.append(f" {result.description}") lines.append(f"{'─' * 80}") # Token metrics lines.append(f" Tokens (original): {result.tokens_original:>12,}") lines.append(f" Tokens (optimized): {result.tokens_optimized:>12,}") lines.append(f" Compression: {result.compression_ratio * 100:>11.1f}%") # Cache metrics (if applicable) if result.cache_hit_rate_optimized > 0: lines.append(f" Cache Hit (before): {result.cache_hit_rate_baseline * 100:>11.1f}%") lines.append(f" Cache Hit (after): {result.cache_hit_rate_optimized * 100:>11.1f}%") # Quality metrics (if applicable) if result.critical_items_total > 0: lines.append( f" Critical Items: {result.critical_items_retained}/{result.critical_items_total} retained" ) lines.append(f" Retention Rate: {result.retention_rate * 100:>11.1f}%") # Cost analysis ca = result.cost_analysis if ca.cost_baseline > 0: lines.append(f" Cost (baseline): ${ca.cost_baseline:>11.4f}") if ca.cost_optimized > 0: lines.append(f" Cost (optimized): ${ca.cost_optimized:>11.4f}") if ca.cost_with_cache > 0: lines.append(f" Cost (with cache): ${ca.cost_with_cache:>11.4f}") lines.append(f" Savings: {ca.total_savings_percent:>11.1f}%") total_baseline += ca.cost_baseline if ca.cost_optimized > 0: total_savings += ca.cost_baseline - ca.cost_optimized elif ca.cost_with_cache > 0: total_savings += ca.cost_baseline - ca.cost_with_cache # Performance if result.optimization_latency_ms > 0: lines.append(f" Optimization Time: {result.optimization_latency_ms:>11.2f}ms") # Summary lines.append("") lines.append("=" * 80) lines.append(" SUMMARY") lines.append("=" * 80) if total_baseline > 0: lines.append(f" Total Baseline Cost: ${total_baseline:.4f}") lines.append(f" Total Savings: ${total_savings:.4f}") lines.append(f" Overall Reduction: {(total_savings / total_baseline) * 100:.1f}%") lines.append("") lines.append(" At 1M requests/month:") lines.append(f" Without Headroom: ${total_baseline * 1_000_000:.2f}") lines.append(f" With Headroom: ${(total_baseline - total_savings) * 1_000_000:.2f}") lines.append(f" Monthly Savings: ${total_savings * 1_000_000:.2f}") lines.append("") return "\n".join(lines) def _generate_markdown_report(results: list[BenchmarkResult]) -> str: """Generate markdown report for documentation.""" lines = [] lines.append("# Headroom Agent Cost Benchmark") lines.append("") lines.append("> The Context Optimization Layer for LLM Applications") lines.append("") lines.append("## Executive Summary") lines.append("") lines.append("This benchmark demonstrates Headroom's impact on real-world agent workloads:") lines.append("") lines.append("| Metric | Impact |") lines.append("|--------|--------|") # Calculate summary metrics total_compression = statistics.mean( [r.compression_ratio for r in results if r.compression_ratio > 0] ) cache_improvement = next((r for r in results if r.cache_hit_rate_optimized > 0), None) quality_result = next((r for r in results if r.retention_rate > 0), None) lines.append(f"| Token Reduction | **{total_compression * 100:.0f}%** average compression |") if cache_improvement: lines.append( f"| Cache Hit Rate | **{cache_improvement.cache_hit_rate_baseline * 100:.0f}% → {cache_improvement.cache_hit_rate_optimized * 100:.0f}%** |" ) if quality_result: lines.append( f"| Quality Retention | **{quality_result.retention_rate * 100:.0f}%** critical items preserved |" ) lines.append("") # Detailed results lines.append("## Detailed Results") lines.append("") for result in results: lines.append(f"### {result.name}") lines.append("") lines.append(f"*{result.description}*") lines.append("") lines.append("| Metric | Value |") lines.append("|--------|-------|") lines.append(f"| Original Tokens | {result.tokens_original:,} |") lines.append(f"| Optimized Tokens | {result.tokens_optimized:,} |") lines.append(f"| Compression | {result.compression_ratio * 100:.1f}% |") if result.cost_analysis.total_savings_percent > 0: lines.append(f"| Cost Savings | {result.cost_analysis.total_savings_percent:.1f}% |") if result.retention_rate > 0: lines.append(f"| Quality Retention | {result.retention_rate * 100:.1f}% |") lines.append("") # Cost projection lines.append("## Cost Projection at Scale") lines.append("") lines.append("Based on Claude 3.5 Sonnet pricing ($3/1M input tokens):") lines.append("") lines.append("| Scale | Without Headroom | With Headroom | Monthly Savings |") lines.append("|-------|------------------|---------------|-----------------|") base_cost_per_request = sum(r.cost_analysis.cost_baseline for r in results) / len(results) optimized_cost = sum( r.cost_analysis.cost_optimized or r.cost_analysis.cost_with_cache or r.cost_analysis.cost_baseline * 0.5 for r in results ) / len(results) for scale, label in [(10_000, "10K"), (100_000, "100K"), (1_000_000, "1M")]: baseline = base_cost_per_request * scale optimized = optimized_cost * scale savings = baseline - optimized lines.append( f"| {label} requests/mo | ${baseline:,.0f} | ${optimized:,.0f} | ${savings:,.0f} |" ) lines.append("") return "\n".join(lines) # ============================================================================= # MAIN # ============================================================================= def main(): parser = argparse.ArgumentParser(description="Headroom Agent Cost Benchmark") parser.add_argument("--format", choices=["terminal", "markdown"], default="terminal") parser.add_argument( "--scenario", choices=["all", "coding-agent", "cache", "rag", "scaling", "quality"], default="all", ) args = parser.parse_args() results = [] print("Running benchmarks...\n") if args.scenario in ("all", "coding-agent"): print(" [1/5] Coding Agent Context Explosion...") results.append(benchmark_coding_agent_explosion()) if args.scenario in ("all", "cache"): print(" [2/5] Cache Alignment Impact...") results.append(benchmark_cache_alignment()) if args.scenario in ("all", "rag"): print(" [3/5] RAG Context Scaling...") results.append(benchmark_rag_scaling()) if args.scenario in ("all", "scaling"): print(" [4/5] Conversation Scaling...") scaling_results = benchmark_conversation_scaling() # Just add the 100-turn result to main results results.append(scaling_results[3]) # 100 turns if args.scenario in ("all", "quality"): print(" [5/5] Quality Preservation...") results.append(benchmark_quality_preservation()) print("\n" + generate_report(results, args.format)) if __name__ == "__main__": main()