mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
e4a41faa33
- Fix E402: Move module-level imports to top of file - Fix F401: Add noqa for availability check imports - Fix F402: Rename loop variables shadowing imports - Fix E722: Replace bare except with except Exception - Fix B904: Add exception chaining (from e) - Fix F811: Remove duplicate imports - Fix B027: Add noqa for empty close() method - Fix E741: Rename ambiguous variable l -> label - Fix I001: Import sorting issues - Apply ruff format to all 106 files All 902 tests pass.
104 lines
3.5 KiB
Python
104 lines
3.5 KiB
Python
"""Verify that SmartCrusher preserves ERROR entries.
|
|
|
|
This is critical - errors should NEVER be dropped during compression.
|
|
"""
|
|
|
|
import json
|
|
|
|
from headroom.config import SmartCrusherConfig
|
|
from headroom.providers import OpenAIProvider
|
|
from headroom.transforms import SmartCrusher
|
|
|
|
from .mock_tools import generate_log_entries
|
|
|
|
|
|
def main():
|
|
print("\n" + "=" * 70)
|
|
print("VERIFYING ERROR PRESERVATION IN SMARTCRUSHER")
|
|
print("=" * 70)
|
|
|
|
# Generate logs with some ERROR entries
|
|
raw_output = generate_log_entries("test-service", count=200)
|
|
data = json.loads(raw_output)
|
|
|
|
# Count errors in original
|
|
original_errors = [e for e in data["entries"] if e["level"] == "ERROR"]
|
|
print(f"\nOriginal log entries: {len(data['entries'])}")
|
|
print(f"ERROR entries in original: {len(original_errors)}")
|
|
print("\nERROR messages found:")
|
|
for err in original_errors[:5]: # Show first 5
|
|
print(f" - {err['message'][:60]}...")
|
|
|
|
# Apply SmartCrusher
|
|
smart_config = SmartCrusherConfig(
|
|
enabled=True,
|
|
min_tokens_to_crush=200,
|
|
max_items_after_crush=20,
|
|
)
|
|
|
|
provider = OpenAIProvider()
|
|
tokenizer = provider.get_token_counter("gpt-4o")
|
|
crusher = SmartCrusher(config=smart_config)
|
|
|
|
messages = [
|
|
{"role": "system", "content": "You are a helpful assistant."},
|
|
{"role": "user", "content": "Find ERROR entries in the logs"},
|
|
{
|
|
"role": "assistant",
|
|
"content": None,
|
|
"tool_calls": [
|
|
{"id": "call_1", "function": {"name": "search_logs", "arguments": "{}"}}
|
|
],
|
|
},
|
|
{"role": "tool", "content": raw_output, "tool_call_id": "call_1"},
|
|
]
|
|
|
|
result = crusher.apply(messages, tokenizer=tokenizer)
|
|
compressed_output = result.messages[-1]["content"]
|
|
|
|
# Handle case where SmartCrusher may add markers
|
|
# Try to find the JSON part
|
|
try:
|
|
compressed_data = json.loads(compressed_output)
|
|
except json.JSONDecodeError:
|
|
# Try to extract just the JSON object
|
|
import re
|
|
|
|
json_match = re.search(r"(\{.*\})", compressed_output, re.DOTALL)
|
|
if json_match:
|
|
compressed_data = json.loads(json_match.group(1))
|
|
else:
|
|
print("Could not parse compressed output:")
|
|
print(compressed_output[:500])
|
|
return
|
|
|
|
# Count errors in compressed
|
|
compressed_errors = [e for e in compressed_data["entries"] if e["level"] == "ERROR"]
|
|
print(f"\nCompressed log entries: {len(compressed_data['entries'])}")
|
|
print(f"ERROR entries preserved: {len(compressed_errors)}")
|
|
print("\nERROR messages in compressed:")
|
|
for err in compressed_errors[:5]:
|
|
print(f" - {err['message'][:60]}...")
|
|
|
|
# Verification
|
|
print("\n" + "=" * 70)
|
|
if len(compressed_errors) >= len(original_errors):
|
|
print("SUCCESS: All ERROR entries were preserved!")
|
|
elif len(compressed_errors) > 0:
|
|
print(f"PARTIAL: {len(compressed_errors)}/{len(original_errors)} ERROR entries preserved")
|
|
else:
|
|
print("FAILURE: ERROR entries were dropped!")
|
|
print("=" * 70)
|
|
|
|
# Show compression ratio
|
|
original_count = len(data["entries"])
|
|
compressed_count = len(compressed_data["entries"])
|
|
reduction = (original_count - compressed_count) / original_count * 100
|
|
print(
|
|
f"\nCompression: {original_count} → {compressed_count} entries ({reduction:.1f}% reduction)"
|
|
)
|
|
print(f"But kept: {len(compressed_errors)} of {len(original_errors)} ERROR entries")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|