mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
8766d83f68
- Replace marketing claims with actual API test results - Add needle-in-haystack test: critical error at position 67/100 - Show real JSON compression: 100 entries → 6 entries (93.9% reduction) - Verified with Claude Sonnet: 87.6% fewer tokens, 4/4 correct answers - Add example scripts for reproducing the tests
299 lines
9.8 KiB
Python
299 lines
9.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Needle-in-Haystack Test: Proving Headroom Preserves Critical Information
|
|
|
|
This test embeds a SPECIFIC answer deep in a large JSON response,
|
|
then verifies both baseline and Headroom-optimized calls can find it.
|
|
|
|
The scenario: Production incident logs with ONE critical error buried in the middle.
|
|
The question: "What caused the outage and what's the fix?"
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
|
|
import httpx
|
|
|
|
ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY")
|
|
if not ANTHROPIC_API_KEY:
|
|
raise ValueError("ANTHROPIC_API_KEY environment variable required")
|
|
|
|
# =============================================================================
|
|
# THE NEEDLE: A critical error buried at position 67 out of 100
|
|
# =============================================================================
|
|
CRITICAL_ERROR = {
|
|
"timestamp": "2024-12-15T03:47:23Z",
|
|
"level": "FATAL",
|
|
"service": "payment-gateway",
|
|
"message": "Connection pool exhausted - max_connections=100 exceeded",
|
|
"error_code": "PG-5523",
|
|
"stack_trace": "ConnectionPoolError at PaymentService.processTransaction()",
|
|
"resolution": "Increase max_connections to 500 in config/database.yml",
|
|
"affected_transactions": 1847,
|
|
"incident_id": "INC-2024-1215-001",
|
|
}
|
|
|
|
|
|
# =============================================================================
|
|
# THE HAYSTACK: 99 normal log entries
|
|
# =============================================================================
|
|
def create_log_entries(n: int = 100, needle_position: int = 67) -> list:
|
|
"""Create n log entries with the critical error at needle_position."""
|
|
logs = []
|
|
for i in range(n):
|
|
if i == needle_position:
|
|
# Insert the needle
|
|
logs.append(CRITICAL_ERROR)
|
|
else:
|
|
# Normal log entry
|
|
logs.append(
|
|
{
|
|
"timestamp": f"2024-12-15T{(i % 24):02d}:{(i % 60):02d}:00Z",
|
|
"level": "INFO",
|
|
"service": ["api-gateway", "user-service", "inventory", "auth"][i % 4],
|
|
"message": f"Request processed successfully - latency={50 + (i % 100)}ms",
|
|
"request_id": f"req-{i:06d}",
|
|
"status_code": 200,
|
|
"endpoint": ["/api/users", "/api/products", "/api/orders", "/health"][i % 4],
|
|
"metadata": {
|
|
"region": ["us-east-1", "us-west-2", "eu-west-1"][i % 3],
|
|
"version": "2.4.1",
|
|
},
|
|
}
|
|
)
|
|
return logs
|
|
|
|
|
|
# =============================================================================
|
|
# THE QUESTION
|
|
# =============================================================================
|
|
QUESTION = """Based on these production logs, answer these specific questions:
|
|
|
|
1. What service caused the outage?
|
|
2. What was the exact error code?
|
|
3. What is the specific fix mentioned in the logs?
|
|
4. How many transactions were affected?
|
|
|
|
Be precise - cite the exact values from the logs."""
|
|
|
|
|
|
def create_messages(tool_content: str) -> list:
|
|
"""Create the conversation with tool output."""
|
|
return [
|
|
{
|
|
"role": "user",
|
|
"content": "Search the production logs for the root cause of last night's outage",
|
|
},
|
|
{
|
|
"role": "assistant",
|
|
"content": [
|
|
{"type": "text", "text": "I'll search the production logs for errors."},
|
|
{
|
|
"type": "tool_use",
|
|
"id": "logs_1",
|
|
"name": "search_logs",
|
|
"input": {"query": "error OR fatal", "limit": 100},
|
|
},
|
|
],
|
|
},
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{
|
|
"type": "tool_result",
|
|
"tool_use_id": "logs_1",
|
|
"content": tool_content,
|
|
}
|
|
],
|
|
},
|
|
{
|
|
"role": "user",
|
|
"content": QUESTION,
|
|
},
|
|
]
|
|
|
|
|
|
def make_api_call(base_url: str, messages: list, label: str) -> dict:
|
|
"""Make API call and return response + usage."""
|
|
headers = {
|
|
"x-api-key": ANTHROPIC_API_KEY,
|
|
"anthropic-version": "2023-06-01",
|
|
"content-type": "application/json",
|
|
}
|
|
|
|
payload = {
|
|
"model": "claude-sonnet-4-20250514",
|
|
"max_tokens": 1000,
|
|
"messages": messages,
|
|
"tools": [
|
|
{
|
|
"name": "search_logs",
|
|
"description": "Search production logs",
|
|
"input_schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"query": {"type": "string"},
|
|
"limit": {"type": "integer"},
|
|
},
|
|
"required": ["query"],
|
|
},
|
|
}
|
|
],
|
|
}
|
|
|
|
print(f"\n{'=' * 70}")
|
|
print(f"{label}")
|
|
print(f"{'=' * 70}")
|
|
|
|
try:
|
|
with httpx.Client(timeout=120.0) as client:
|
|
response = client.post(
|
|
f"{base_url}/v1/messages",
|
|
headers=headers,
|
|
json=payload,
|
|
)
|
|
|
|
if response.status_code != 200:
|
|
print(f"Error: {response.status_code}")
|
|
print(response.text[:500])
|
|
return {"error": response.text}
|
|
|
|
data = response.json()
|
|
usage = data.get("usage", {})
|
|
|
|
# Extract text response
|
|
response_text = ""
|
|
for block in data.get("content", []):
|
|
if block.get("type") == "text":
|
|
response_text += block.get("text", "")
|
|
|
|
return {
|
|
"input_tokens": usage.get("input_tokens", 0),
|
|
"output_tokens": usage.get("output_tokens", 0),
|
|
"response": response_text,
|
|
}
|
|
|
|
except Exception as e:
|
|
print(f"Exception: {e}")
|
|
return {"error": str(e)}
|
|
|
|
|
|
def verify_answer(response: str) -> dict:
|
|
"""Check if the response contains the correct answers."""
|
|
checks = {
|
|
"service": "payment-gateway" in response.lower(),
|
|
"error_code": "PG-5523" in response or "pg-5523" in response.lower(),
|
|
"fix": "max_connections" in response.lower() or "500" in response,
|
|
"transactions": "1847" in response or "1,847" in response,
|
|
}
|
|
return checks
|
|
|
|
|
|
def main():
|
|
print("\n" + "=" * 70)
|
|
print("NEEDLE-IN-HAYSTACK TEST")
|
|
print("Proving Headroom preserves critical information")
|
|
print("=" * 70)
|
|
|
|
# Create the test data
|
|
logs = create_log_entries(100, needle_position=67)
|
|
log_json = json.dumps(logs)
|
|
|
|
print("\nTest setup:")
|
|
print(f" - Total log entries: {len(logs)}")
|
|
print(" - Critical error at position: 67")
|
|
print(f" - JSON size: {len(log_json):,} characters")
|
|
|
|
print("\nThe needle (what we're looking for):")
|
|
print(f" - Service: {CRITICAL_ERROR['service']}")
|
|
print(f" - Error code: {CRITICAL_ERROR['error_code']}")
|
|
print(f" - Fix: {CRITICAL_ERROR['resolution']}")
|
|
print(f" - Affected: {CRITICAL_ERROR['affected_transactions']} transactions")
|
|
|
|
messages = create_messages(log_json)
|
|
|
|
# Test 1: Baseline (direct to Anthropic)
|
|
print("\n" + "-" * 70)
|
|
baseline = make_api_call(
|
|
"https://api.anthropic.com",
|
|
messages,
|
|
"BASELINE: Direct to Anthropic API",
|
|
)
|
|
|
|
if "error" not in baseline:
|
|
print(f"\nInput tokens: {baseline['input_tokens']:,}")
|
|
print(f"\nResponse:\n{baseline['response'][:1000]}...")
|
|
baseline_checks = verify_answer(baseline["response"])
|
|
print(f"\nAnswer verification: {baseline_checks}")
|
|
|
|
# Test 2: Headroom optimized
|
|
print("\n" + "-" * 70)
|
|
optimized = make_api_call(
|
|
"http://localhost:8787",
|
|
messages,
|
|
"HEADROOM: Through optimization proxy",
|
|
)
|
|
|
|
if "error" not in optimized:
|
|
print(f"\nInput tokens: {optimized['input_tokens']:,}")
|
|
print(f"\nResponse:\n{optimized['response'][:1000]}...")
|
|
optimized_checks = verify_answer(optimized["response"])
|
|
print(f"\nAnswer verification: {optimized_checks}")
|
|
|
|
# Final comparison
|
|
print("\n" + "=" * 70)
|
|
print("FINAL RESULTS")
|
|
print("=" * 70)
|
|
|
|
if "error" not in baseline and "error" not in optimized:
|
|
baseline_input = baseline["input_tokens"]
|
|
optimized_input = optimized["input_tokens"]
|
|
saved = baseline_input - optimized_input
|
|
percent = (saved / baseline_input * 100) if baseline_input > 0 else 0
|
|
|
|
baseline_score = sum(baseline_checks.values())
|
|
optimized_score = sum(optimized_checks.values())
|
|
|
|
print(f"""
|
|
TOKEN COMPARISON:
|
|
Baseline: {baseline_input:,} tokens
|
|
Headroom: {optimized_input:,} tokens
|
|
Saved: {saved:,} tokens ({percent:.1f}% reduction)
|
|
|
|
ANSWER ACCURACY (4 questions):
|
|
Baseline: {baseline_score}/4 correct
|
|
Headroom: {optimized_score}/4 correct
|
|
|
|
VERIFICATION DETAILS:
|
|
Baseline: {baseline_checks}
|
|
Headroom: {optimized_checks}
|
|
|
|
CONCLUSION:
|
|
{"PASS - Headroom found the needle!" if optimized_score >= 3 else "NEEDS REVIEW"}
|
|
{"Answer quality maintained with " + f"{percent:.0f}% fewer tokens" if optimized_score >= baseline_score else ""}
|
|
""")
|
|
|
|
return {
|
|
"baseline_tokens": baseline_input,
|
|
"optimized_tokens": optimized_input,
|
|
"tokens_saved": saved,
|
|
"percent_saved": percent,
|
|
"baseline_accuracy": baseline_score,
|
|
"optimized_accuracy": optimized_score,
|
|
"baseline_checks": baseline_checks,
|
|
"optimized_checks": optimized_checks,
|
|
"baseline_response": baseline["response"],
|
|
"optimized_response": optimized["response"],
|
|
}
|
|
|
|
return None
|
|
|
|
|
|
if __name__ == "__main__":
|
|
result = main()
|
|
if result:
|
|
print("\n" + "=" * 70)
|
|
print("JSON SUMMARY:")
|
|
print("=" * 70)
|
|
summary = {k: v for k, v in result.items() if not k.endswith("_response")}
|
|
print(json.dumps(summary, indent=2))
|