mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
Reduce proxy latency, enrich telemetry, protect Bash output, bump to 0.5.6
Performance: - Replace json roundtrip with copy.deepcopy in deep_copy_messages (~20-80ms) - Eliminate redundant token counting in pipeline (8→2 calls, ~30-100ms) - Parallel message compression in ContentRouter via ThreadPoolExecutor (~100-200ms) - Add granular timing metrics: deep_copy, token_count, parallel_compress - Switch hot-path hashing from SHA256 to MD5 (2-3x faster, non-crypto) Telemetry: - Enrich beacon payload with overhead, TTFB, pipeline timing, request patterns, compression cache stats, CCR usage, and waste signals - Each extraction section guarded independently so one bad key never blocks the rest; Supabase POST failure never affects proxy Bug fix: - Add Bash/bash to DEFAULT_EXCLUDE_TOOLS so tree/ls output is not compressed by the text compressor
This commit is contained in:
parent
f1338093c0
commit
4605fc1971
14 changed files with 288 additions and 81 deletions
|
|
@ -153,7 +153,7 @@ from .transforms import (
|
|||
TransformPipeline,
|
||||
)
|
||||
|
||||
__version__ = "0.5.5"
|
||||
__version__ = "0.5.6"
|
||||
|
||||
__all__ = [
|
||||
# Main client
|
||||
|
|
|
|||
2
headroom/cache/base.py
vendored
2
headroom/cache/base.py
vendored
|
|
@ -320,7 +320,7 @@ class BaseCacheOptimizer(ABC):
|
|||
"""Compute a short hash of content."""
|
||||
import hashlib
|
||||
|
||||
return hashlib.sha256(content.encode()).hexdigest()[:12]
|
||||
return hashlib.md5(content.encode()).hexdigest()[:12]
|
||||
|
||||
def _extract_system_content(self, messages: list[dict[str, Any]]) -> str:
|
||||
"""Extract content from system messages."""
|
||||
|
|
|
|||
2
headroom/cache/compression_cache.py
vendored
2
headroom/cache/compression_cache.py
vendored
|
|
@ -135,7 +135,7 @@ class CompressionCache:
|
|||
raw = json.dumps(content, sort_keys=True, ensure_ascii=False)
|
||||
else:
|
||||
raw = content
|
||||
return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16]
|
||||
return hashlib.md5(raw.encode("utf-8")).hexdigest()[:16]
|
||||
|
||||
def compute_frozen_count(self, messages: list[dict]) -> int:
|
||||
"""Count consecutive stable messages from the start.
|
||||
|
|
|
|||
2
headroom/cache/compression_store.py
vendored
2
headroom/cache/compression_store.py
vendored
|
|
@ -206,7 +206,7 @@ class CompressionStore:
|
|||
# collision resistance. Birthday paradox: 50% collision at sqrt(2^n) entries.
|
||||
# - 64 bits: ~4 billion entries for 50% collision
|
||||
# - 96 bits: ~280 trillion entries for 50% collision
|
||||
hash_key = hashlib.sha256(original.encode()).hexdigest()[:24]
|
||||
hash_key = hashlib.md5(original.encode()).hexdigest()[:24]
|
||||
|
||||
entry = CompressionEntry(
|
||||
hash=hash_key,
|
||||
|
|
|
|||
2
headroom/cache/prefix_tracker.py
vendored
2
headroom/cache/prefix_tracker.py
vendored
|
|
@ -293,7 +293,7 @@ class SessionTrackerStore:
|
|||
break
|
||||
|
||||
key = f"{model}:{system_content}"
|
||||
return hashlib.sha256(key.encode()).hexdigest()[:16]
|
||||
return hashlib.md5(key.encode()).hexdigest()[:16]
|
||||
|
||||
def _maybe_cleanup(self) -> None:
|
||||
"""Remove expired trackers periodically."""
|
||||
|
|
|
|||
|
|
@ -363,12 +363,14 @@ DEFAULT_EXCLUDE_TOOLS: frozenset[str] = frozenset(
|
|||
"Grep",
|
||||
"Write",
|
||||
"Edit",
|
||||
"Bash",
|
||||
# Lowercase variants for case-insensitive matching
|
||||
"read",
|
||||
"glob",
|
||||
"grep",
|
||||
"write",
|
||||
"edit",
|
||||
"bash",
|
||||
}
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -32,8 +32,8 @@ RAG_PATTERN = re.compile("|".join(RAG_MARKERS), re.IGNORECASE)
|
|||
|
||||
|
||||
def compute_hash(text: str) -> str:
|
||||
"""Compute SHA256 hash of text, truncated to 16 chars."""
|
||||
return hashlib.sha256(text.encode()).hexdigest()[:16]
|
||||
"""Compute hash of text, truncated to 16 chars."""
|
||||
return hashlib.md5(text.encode()).hexdigest()[:16]
|
||||
|
||||
|
||||
def detect_waste_signals(text: str, tokenizer: Tokenizer) -> WasteSignals:
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
"""Anonymous usage telemetry beacon for Headroom.
|
||||
|
||||
Sends aggregate-only stats (tokens saved, compression ratios, cache hit rates)
|
||||
to help improve Headroom. No prompts, no content, no PII.
|
||||
Sends aggregate-only stats (tokens saved, compression ratios, cache hit rates,
|
||||
performance overhead) to help improve Headroom. No prompts, no content, no PII.
|
||||
|
||||
On by default. Opt out with:
|
||||
HEADROOM_TELEMETRY=off headroom proxy
|
||||
|
|
@ -84,13 +84,22 @@ class TelemetryBeacon:
|
|||
await asyncio.sleep(_INTERVAL_SECONDS)
|
||||
|
||||
async def _report(self) -> None:
|
||||
"""Fetch stats from local /stats endpoint and POST to Supabase."""
|
||||
"""Fetch stats from local /stats endpoint and POST to Supabase.
|
||||
|
||||
Wrapped in multiple try/except layers so that:
|
||||
1. A missing httpx import silently skips.
|
||||
2. A failed /stats fetch silently skips.
|
||||
3. Extraction of any stats section is independent — one bad key
|
||||
never blocks the others.
|
||||
4. A failed Supabase POST silently skips (fire-and-forget).
|
||||
The proxy NEVER crashes or slows down because of telemetry.
|
||||
"""
|
||||
try:
|
||||
import httpx
|
||||
except ImportError:
|
||||
return
|
||||
|
||||
# Fetch stats from our own proxy
|
||||
# ---- Fetch stats from our own proxy ----
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=5.0) as client:
|
||||
resp = await client.get(f"http://127.0.0.1:{self._port}/stats")
|
||||
|
|
@ -100,18 +109,15 @@ class TelemetryBeacon:
|
|||
except Exception:
|
||||
return
|
||||
|
||||
tokens = stats.get("tokens", {})
|
||||
requests = stats.get("requests", {})
|
||||
cache = stats.get("prefix_cache", {}).get("totals", {})
|
||||
cost = stats.get("cost", {})
|
||||
models_by = requests.get("by_model", {})
|
||||
models = [m for m in models_by.keys() if not m.startswith("passthrough:")]
|
||||
|
||||
# Don't send empty stats — no point reporting zeros
|
||||
total_requests = requests.get("total", 0)
|
||||
if total_requests == 0:
|
||||
try:
|
||||
total_requests = stats.get("requests", {}).get("total", 0)
|
||||
if total_requests == 0:
|
||||
return
|
||||
except Exception:
|
||||
return
|
||||
|
||||
# ---- Build payload — each section guarded independently ----
|
||||
session_minutes = max(1, int((time.time() - self._start_time) / 60))
|
||||
|
||||
try:
|
||||
|
|
@ -119,24 +125,126 @@ class TelemetryBeacon:
|
|||
except Exception:
|
||||
headroom_version = "unknown"
|
||||
|
||||
payload = {
|
||||
# Core identity (always present)
|
||||
payload: dict = {
|
||||
"session_id": self._session_id,
|
||||
"instance_id": self._instance_id,
|
||||
"headroom_version": headroom_version,
|
||||
"python_version": f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}",
|
||||
"python_version": (
|
||||
f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}"
|
||||
),
|
||||
"os": f"{platform.system()} {platform.machine()}",
|
||||
"sdk": self._sdk,
|
||||
"backend": self._backend,
|
||||
"tokens_saved": tokens.get("saved", 0),
|
||||
"requests": requests.get("total", 0),
|
||||
"compression_percent": tokens.get("savings_percent", 0),
|
||||
"cache_hit_rate": cache.get("hit_rate", 0),
|
||||
"cost_saved_usd": cost.get("savings_usd", 0),
|
||||
"cache_saved_usd": cache.get("cache_savings_usd", 0),
|
||||
"session_minutes": session_minutes,
|
||||
"models_used": models,
|
||||
}
|
||||
|
||||
# --- Effectiveness metrics ---
|
||||
try:
|
||||
tokens = stats.get("tokens", {})
|
||||
requests_stats = stats.get("requests", {})
|
||||
cache = stats.get("prefix_cache", {}).get("totals", {})
|
||||
cost = stats.get("cost", {})
|
||||
models_by = requests_stats.get("by_model", {})
|
||||
|
||||
payload.update(
|
||||
{
|
||||
"tokens_saved": tokens.get("saved", 0),
|
||||
"requests": requests_stats.get("total", 0),
|
||||
"compression_percent": tokens.get("savings_percent", 0),
|
||||
"cache_hit_rate": cache.get("hit_rate", 0),
|
||||
"cost_saved_usd": cost.get("savings_usd", 0),
|
||||
"cache_saved_usd": cost.get("cache_savings_usd", 0),
|
||||
"models_used": [
|
||||
m for m in models_by.keys() if not m.startswith("passthrough:")
|
||||
],
|
||||
}
|
||||
)
|
||||
except Exception:
|
||||
logger.debug("Beacon: failed to extract effectiveness metrics", exc_info=True)
|
||||
|
||||
# --- Performance overhead (how much latency Headroom adds) ---
|
||||
try:
|
||||
overhead = stats.get("overhead", {})
|
||||
payload.update(
|
||||
{
|
||||
"overhead_avg_ms": round(overhead.get("average_ms", 0), 2),
|
||||
"overhead_max_ms": round(overhead.get("max_ms", 0), 2),
|
||||
}
|
||||
)
|
||||
except Exception:
|
||||
logger.debug("Beacon: failed to extract overhead metrics", exc_info=True)
|
||||
|
||||
# --- TTFB (time to first byte — what the user feels) ---
|
||||
try:
|
||||
ttfb = stats.get("ttfb", {})
|
||||
payload["ttfb_avg_ms"] = round(ttfb.get("average_ms", 0), 2)
|
||||
except Exception:
|
||||
logger.debug("Beacon: failed to extract TTFB metrics", exc_info=True)
|
||||
|
||||
# --- Pipeline timing breakdown (where is time spent?) ---
|
||||
# Stored as JSONB — variable-shape dict of transform_name → avg_ms.
|
||||
# This is the most valuable data for optimising Headroom internals.
|
||||
try:
|
||||
raw_timing = stats.get("pipeline_timing", {})
|
||||
if raw_timing:
|
||||
# Flatten to {name: avg_ms} for compact storage
|
||||
pipeline_timing = {
|
||||
name: round(info.get("average_ms", 0), 2)
|
||||
for name, info in raw_timing.items()
|
||||
if isinstance(info, dict)
|
||||
}
|
||||
payload["pipeline_timing"] = pipeline_timing
|
||||
except Exception:
|
||||
logger.debug("Beacon: failed to extract pipeline timing", exc_info=True)
|
||||
|
||||
# --- Request patterns (how big are conversations?) ---
|
||||
try:
|
||||
tokens = stats.get("tokens", {})
|
||||
total_req = stats.get("requests", {}).get("total", 1)
|
||||
tokens_before = tokens.get("original", 0)
|
||||
tokens_after = tokens.get("optimized", 0)
|
||||
payload.update(
|
||||
{
|
||||
"avg_tokens_before": round(tokens_before / max(total_req, 1)),
|
||||
"avg_tokens_after": round(tokens_after / max(total_req, 1)),
|
||||
}
|
||||
)
|
||||
except Exception:
|
||||
logger.debug("Beacon: failed to extract request patterns", exc_info=True)
|
||||
|
||||
# --- Compression cache effectiveness ---
|
||||
try:
|
||||
cc = stats.get("compression_cache", {})
|
||||
if cc:
|
||||
payload["compression_cache"] = {
|
||||
"hit_rate": cc.get("hit_rate", 0),
|
||||
"entries": cc.get("entries", 0),
|
||||
"avg_lookup_ns": cc.get("avg_lookup_ns", 0),
|
||||
}
|
||||
except Exception:
|
||||
logger.debug("Beacon: failed to extract cache stats", exc_info=True)
|
||||
|
||||
# --- CCR (Compress-Cache-Retrieve) usage ---
|
||||
try:
|
||||
ccr = stats.get("compression", {})
|
||||
if ccr.get("ccr_entries", 0) > 0:
|
||||
payload["ccr"] = {
|
||||
"entries": ccr.get("ccr_entries", 0),
|
||||
"retrievals": ccr.get("ccr_retrievals", 0),
|
||||
}
|
||||
except Exception:
|
||||
logger.debug("Beacon: failed to extract CCR stats", exc_info=True)
|
||||
|
||||
# --- Waste signals (what patterns of waste do we see?) ---
|
||||
try:
|
||||
waste = stats.get("waste_signals", {})
|
||||
if waste:
|
||||
payload["waste_signals"] = waste
|
||||
except Exception:
|
||||
logger.debug("Beacon: failed to extract waste signals", exc_info=True)
|
||||
|
||||
# ---- Send to Supabase (fire-and-forget) ----
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=10.0) as client:
|
||||
await client.post(
|
||||
|
|
@ -151,4 +259,6 @@ class TelemetryBeacon:
|
|||
},
|
||||
)
|
||||
except Exception:
|
||||
pass # Fire and forget
|
||||
# No internet, DNS failure, timeout, Supabase down — all fine.
|
||||
# Headroom continues working perfectly without telemetry.
|
||||
logger.debug("Beacon: failed to send telemetry", exc_info=True)
|
||||
|
|
|
|||
|
|
@ -334,7 +334,7 @@ def compute_item_hash(item: dict[str, Any]) -> str:
|
|||
content = json.dumps(item, sort_keys=True, default=str)
|
||||
except (TypeError, ValueError):
|
||||
content = str(item)
|
||||
return hashlib.sha256(content.encode()).hexdigest()[:16]
|
||||
return hashlib.md5(content.encode()).hexdigest()[:16]
|
||||
|
||||
|
||||
class AnchorSelector:
|
||||
|
|
|
|||
|
|
@ -37,8 +37,10 @@ from __future__ import annotations
|
|||
|
||||
import hashlib
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import time
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from dataclasses import dataclass, field
|
||||
from enum import Enum
|
||||
from typing import Any
|
||||
|
|
@ -120,6 +122,8 @@ def _create_content_signature(
|
|||
structure_hint = hashlib.sha256(content_sample.encode()).hexdigest()[:8]
|
||||
hash_input = f"{hash_input}:{structure_hint}"
|
||||
|
||||
# Keep SHA256: structure_hash feeds into TOIN which persists to disk.
|
||||
# Changing hash function would invalidate all learned patterns.
|
||||
structure_hash = hashlib.sha256(hash_input.encode()).hexdigest()[:24]
|
||||
|
||||
return ToolSignature(
|
||||
|
|
@ -717,6 +721,14 @@ class ContentRouter(Transform):
|
|||
# TOIN recording should never break compression
|
||||
logger.debug("TOIN recording failed (non-fatal): %s", e)
|
||||
|
||||
def _timed_compress(
|
||||
self, content: str, context: str, bias: float
|
||||
) -> tuple[RouterCompressionResult, float]:
|
||||
"""Compress with wall-clock timing. Used by parallel executor."""
|
||||
t0 = time.perf_counter()
|
||||
result = self.compress(content, context=context, bias=bias)
|
||||
return result, (time.perf_counter() - t0) * 1000
|
||||
|
||||
def compress(
|
||||
self,
|
||||
content: str,
|
||||
|
|
@ -1514,12 +1526,33 @@ class ContentRouter(Transform):
|
|||
|
||||
frozen_message_count = kwargs.get("frozen_message_count", 0)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Two-pass parallel compression.
|
||||
#
|
||||
# Pass 1 (sequential): categorise every message — frozen, protected,
|
||||
# cached, small, etc. are resolved immediately. Cache-miss messages
|
||||
# that need full compression are collected into *pending_tasks*.
|
||||
#
|
||||
# Pass 2 (parallel): all cache-miss compressions run concurrently in
|
||||
# a thread pool. Each self.compress() call is independent.
|
||||
#
|
||||
# Pass 3 (sequential): results are stitched back into message order,
|
||||
# caches updated, and counters incremented.
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
# Pre-allocate result slots — None means "pending compression".
|
||||
result_slots: list[dict[str, Any] | None] = [None] * num_messages
|
||||
|
||||
# Tasks: list of (slot_index, content, context, bias, content_key)
|
||||
_PendingTask = tuple[int, str, str, float, int]
|
||||
pending_tasks: list[_PendingTask] = []
|
||||
|
||||
for i, message in enumerate(messages):
|
||||
# Skip frozen messages (in provider's prefix cache).
|
||||
# Modifying these would invalidate the cache, replacing a 90%
|
||||
# read discount with a 25% write penalty (Anthropic).
|
||||
if i < frozen_message_count:
|
||||
transformed_messages.append(message)
|
||||
result_slots[i] = message
|
||||
continue
|
||||
|
||||
role = message.get("role", "")
|
||||
|
|
@ -1544,13 +1577,13 @@ class ContentRouter(Transform):
|
|||
messages_from_end=messages_from_end,
|
||||
compressor_timing=compressor_timing,
|
||||
)
|
||||
transformed_messages.append(transformed_message)
|
||||
result_slots[i] = transformed_message
|
||||
route_counts["content_blocks"] += 1
|
||||
continue
|
||||
|
||||
# Skip non-string content (other types)
|
||||
if not isinstance(content, str):
|
||||
transformed_messages.append(message)
|
||||
result_slots[i] = message
|
||||
route_counts["non_string"] += 1
|
||||
continue
|
||||
|
||||
|
|
@ -1562,7 +1595,7 @@ class ContentRouter(Transform):
|
|||
if tool_call_id in excluded_tool_ids:
|
||||
if messages_from_end <= read_protection_window:
|
||||
# Recent — protect as before
|
||||
transformed_messages.append(message)
|
||||
result_slots[i] = message
|
||||
transforms_applied.append("router:excluded:tool")
|
||||
route_counts["excluded_tool"] += 1
|
||||
continue
|
||||
|
|
@ -1575,14 +1608,14 @@ class ContentRouter(Transform):
|
|||
|
||||
# Protection 1: Never compress user messages
|
||||
if self.config.skip_user_messages and role == "user":
|
||||
transformed_messages.append(message)
|
||||
result_slots[i] = message
|
||||
transforms_applied.append("router:protected:user_message")
|
||||
route_counts["user_msg"] += 1
|
||||
continue
|
||||
|
||||
if not content or len(content.split()) < 50:
|
||||
# Skip small content
|
||||
transformed_messages.append(message)
|
||||
result_slots[i] = message
|
||||
route_counts["small"] += 1
|
||||
continue
|
||||
|
||||
|
|
@ -1597,14 +1630,14 @@ class ContentRouter(Transform):
|
|||
and messages_from_end <= self.config.protect_recent_code
|
||||
and is_code
|
||||
):
|
||||
transformed_messages.append(message)
|
||||
result_slots[i] = message
|
||||
transforms_applied.append("router:protected:recent_code")
|
||||
route_counts["recent_code"] += 1
|
||||
continue
|
||||
|
||||
# Protection 3: Don't compress CODE when analysis intent detected
|
||||
if analysis_intent and is_code:
|
||||
transformed_messages.append(message)
|
||||
result_slots[i] = message
|
||||
transforms_applied.append("router:protected:analysis_context")
|
||||
route_counts["analysis_ctx"] += 1
|
||||
continue
|
||||
|
|
@ -1614,7 +1647,7 @@ class ContentRouter(Transform):
|
|||
# Recompressing would change byte content and break provider
|
||||
# prefix caching with no meaningful further reduction.
|
||||
if "Retrieve more: hash=" in content or "Retrieve original: hash=" in content:
|
||||
transformed_messages.append(message)
|
||||
result_slots[i] = message
|
||||
route_counts.setdefault("already_compressed", 0)
|
||||
route_counts["already_compressed"] += 1
|
||||
continue
|
||||
|
|
@ -1632,7 +1665,7 @@ class ContentRouter(Transform):
|
|||
|
||||
# Tier 1: skip set — instant rejection
|
||||
if self._cache.is_skipped(content_key):
|
||||
transformed_messages.append(message)
|
||||
result_slots[i] = message
|
||||
route_counts["ratio_too_high"] += 1
|
||||
route_counts.setdefault("cache_hit", 0)
|
||||
route_counts["cache_hit"] += 1
|
||||
|
|
@ -1644,47 +1677,83 @@ class ContentRouter(Transform):
|
|||
cached_compressed, cached_ratio, cached_strategy = cached
|
||||
# Re-check ratio against current min_ratio (shifts with context pressure)
|
||||
if cached_ratio < min_ratio:
|
||||
transformed_messages.append({**message, "content": cached_compressed})
|
||||
result_slots[i] = {**message, "content": cached_compressed}
|
||||
transforms_applied.append(f"router:{cached_strategy}:{cached_ratio:.2f}")
|
||||
compressed_details.append(f"{cached_strategy}:{cached_ratio:.2f}")
|
||||
else:
|
||||
# Threshold tightened — no longer qualifies. Move to skip.
|
||||
self._cache.move_to_skip(content_key)
|
||||
transformed_messages.append(message)
|
||||
result_slots[i] = message
|
||||
route_counts["ratio_too_high"] += 1
|
||||
route_counts.setdefault("cache_hit", 0)
|
||||
route_counts["cache_hit"] += 1
|
||||
continue
|
||||
|
||||
# Cache miss — run full compression
|
||||
# Cache miss — defer to parallel compression pass
|
||||
route_counts.setdefault("cache_miss", 0)
|
||||
route_counts["cache_miss"] += 1
|
||||
t0 = time.perf_counter()
|
||||
result = self.compress(content, context=context, bias=msg_bias)
|
||||
compress_ms = (time.perf_counter() - t0) * 1000
|
||||
strategy_key = f"compressor:{result.strategy_used.value}"
|
||||
compressor_timing[strategy_key] = compressor_timing.get(strategy_key, 0.0) + compress_ms
|
||||
pending_tasks.append((i, content, context, msg_bias, content_key))
|
||||
|
||||
if result.compression_ratio < min_ratio:
|
||||
# Compressed — store in result cache
|
||||
self._cache.put(
|
||||
content_key,
|
||||
result.compressed,
|
||||
result.compression_ratio,
|
||||
result.strategy_used.value,
|
||||
)
|
||||
transformed_messages.append({**message, "content": result.compressed})
|
||||
transforms_applied.append(
|
||||
f"router:{result.strategy_used.value}:{result.compression_ratio:.2f}"
|
||||
)
|
||||
compressed_details.append(
|
||||
f"{result.strategy_used.value}:{result.compression_ratio:.2f}"
|
||||
)
|
||||
# --- Pass 2: Parallel compression of all cache-miss messages ---
|
||||
if pending_tasks:
|
||||
max_workers = min(
|
||||
len(pending_tasks), int(os.environ.get("HEADROOM_COMPRESS_WORKERS", "4"))
|
||||
)
|
||||
t_parallel_start = time.perf_counter()
|
||||
|
||||
if max_workers <= 1 or len(pending_tasks) == 1:
|
||||
# Single task or parallelism disabled — compress inline
|
||||
task_results = []
|
||||
for _, task_content, task_ctx, task_bias, _ in pending_tasks:
|
||||
t0 = time.perf_counter()
|
||||
r = self.compress(task_content, context=task_ctx, bias=task_bias)
|
||||
task_results.append((r, (time.perf_counter() - t0) * 1000))
|
||||
else:
|
||||
# Didn't compress — add to skip set
|
||||
self._cache.mark_skip(content_key)
|
||||
transformed_messages.append(message)
|
||||
route_counts["ratio_too_high"] += 1
|
||||
# Parallel compression via thread pool
|
||||
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
||||
futures = []
|
||||
for _, task_content, task_ctx, task_bias, _ in pending_tasks:
|
||||
futures.append(
|
||||
executor.submit(self._timed_compress, task_content, task_ctx, task_bias)
|
||||
)
|
||||
task_results = [f.result() for f in futures]
|
||||
|
||||
parallel_ms = (time.perf_counter() - t_parallel_start) * 1000
|
||||
compressor_timing["parallel_compress_total"] = parallel_ms
|
||||
|
||||
# --- Pass 3: Merge results back (sequential, updates caches) ---
|
||||
for (slot_idx, _, _, _, content_key), (result, compress_ms) in zip(
|
||||
pending_tasks, task_results
|
||||
):
|
||||
message = messages[slot_idx]
|
||||
strategy_key = f"compressor:{result.strategy_used.value}"
|
||||
compressor_timing[strategy_key] = (
|
||||
compressor_timing.get(strategy_key, 0.0) + compress_ms
|
||||
)
|
||||
|
||||
if result.compression_ratio < min_ratio:
|
||||
# Compressed — store in result cache
|
||||
self._cache.put(
|
||||
content_key,
|
||||
result.compressed,
|
||||
result.compression_ratio,
|
||||
result.strategy_used.value,
|
||||
)
|
||||
result_slots[slot_idx] = {**message, "content": result.compressed}
|
||||
transforms_applied.append(
|
||||
f"router:{result.strategy_used.value}:{result.compression_ratio:.2f}"
|
||||
)
|
||||
compressed_details.append(
|
||||
f"{result.strategy_used.value}:{result.compression_ratio:.2f}"
|
||||
)
|
||||
else:
|
||||
# Didn't compress — add to skip set
|
||||
self._cache.mark_skip(content_key)
|
||||
result_slots[slot_idx] = message
|
||||
route_counts["ratio_too_high"] += 1
|
||||
|
||||
# Build final message list from slots
|
||||
transformed_messages = [m for m in result_slots if m is not None]
|
||||
|
||||
tokens_after = sum(
|
||||
tokenizer.count_text(str(m.get("content", ""))) for m in transformed_messages
|
||||
|
|
|
|||
|
|
@ -177,7 +177,9 @@ class TransformPipeline:
|
|||
)
|
||||
|
||||
# Start with original tokens
|
||||
t_count = time.perf_counter()
|
||||
tokens_before = tokenizer.count_messages(messages)
|
||||
count_ms = (time.perf_counter() - t_count) * 1000
|
||||
|
||||
logger.debug(
|
||||
"Pipeline starting: %d messages, %d tokens, model=%s",
|
||||
|
|
@ -196,7 +198,13 @@ class TransformPipeline:
|
|||
transform_diffs: list[TransformDiff] = []
|
||||
generate_diff = self.config.generate_diff_artifact
|
||||
|
||||
t_copy = time.perf_counter()
|
||||
current_messages = deep_copy_messages(messages)
|
||||
copy_ms = (time.perf_counter() - t_copy) * 1000
|
||||
|
||||
all_timing["_deep_copy"] = copy_ms
|
||||
all_timing["_initial_token_count"] = count_ms
|
||||
|
||||
pipeline_start = time.perf_counter()
|
||||
|
||||
frozen_count = kwargs.get("frozen_message_count", 0)
|
||||
|
|
@ -212,9 +220,6 @@ class TransformPipeline:
|
|||
if not transform.should_apply(current_messages, tokenizer, **kwargs):
|
||||
continue
|
||||
|
||||
# Track tokens before this transform (for diff)
|
||||
tokens_before_transform = tokenizer.count_messages(current_messages)
|
||||
|
||||
# Time the transform
|
||||
t0 = time.perf_counter()
|
||||
result = transform.apply(current_messages, tokenizer, **kwargs)
|
||||
|
|
@ -223,8 +228,10 @@ class TransformPipeline:
|
|||
# Update messages for next transform
|
||||
current_messages = result.messages
|
||||
|
||||
# Track tokens after this transform (for diff)
|
||||
tokens_after_transform = tokenizer.count_messages(current_messages)
|
||||
# Use token counts reported by the transform itself — avoids
|
||||
# redundant O(N) recount of the full message list after each step.
|
||||
tokens_before_transform = result.tokens_before
|
||||
tokens_after_transform = result.tokens_after
|
||||
|
||||
# Accumulate results
|
||||
all_transforms.extend(result.transforms_applied)
|
||||
|
|
@ -264,8 +271,12 @@ class TransformPipeline:
|
|||
)
|
||||
)
|
||||
|
||||
# Final token count
|
||||
# Single final token count — the only full recount in the pipeline.
|
||||
# Earlier per-transform counts come from each transform's own result.
|
||||
t_final_count = time.perf_counter()
|
||||
tokens_after = tokenizer.count_messages(current_messages)
|
||||
all_timing["_final_token_count"] = (time.perf_counter() - t_final_count) * 1000
|
||||
|
||||
pipeline_ms = (time.perf_counter() - pipeline_start) * 1000
|
||||
all_timing["pipeline_total"] = pipeline_ms
|
||||
|
||||
|
|
|
|||
|
|
@ -1689,7 +1689,7 @@ class SmartCrusher(Transform):
|
|||
else:
|
||||
# Non-dict items: use string representation
|
||||
content = str(item)
|
||||
item_hash = hashlib.sha256(content.encode()).hexdigest()[:16]
|
||||
item_hash = hashlib.md5(content.encode()).hexdigest()[:16]
|
||||
except (TypeError, ValueError, RecursionError) as e:
|
||||
# Serialization failed - keep the item (fail-safe)
|
||||
logger.debug("Dedup hash failed for item at index %d: %s. Keeping item.", idx, e)
|
||||
|
|
@ -1759,7 +1759,7 @@ class SmartCrusher(Transform):
|
|||
content = json.dumps(item, sort_keys=True, default=str)
|
||||
else:
|
||||
content = str(item)
|
||||
seen_hashes.add(hashlib.sha256(content.encode()).hexdigest()[:16])
|
||||
seen_hashes.add(hashlib.md5(content.encode()).hexdigest()[:16])
|
||||
except (TypeError, ValueError, RecursionError):
|
||||
pass # Skip hash computation failures
|
||||
|
||||
|
|
@ -1792,7 +1792,7 @@ class SmartCrusher(Transform):
|
|||
content = json.dumps(item, sort_keys=True, default=str)
|
||||
else:
|
||||
content = str(item)
|
||||
item_hash = hashlib.sha256(content.encode()).hexdigest()[:16]
|
||||
item_hash = hashlib.md5(content.encode()).hexdigest()[:16]
|
||||
except (TypeError, ValueError, RecursionError):
|
||||
# Hash failure - use index as unique hash (fail-safe)
|
||||
item_hash = f"__idx_{idx}__"
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
|
|
@ -31,6 +32,17 @@ def compute_short_hash(data: str | bytes, length: int = 16) -> str:
|
|||
return compute_hash(data)[:length]
|
||||
|
||||
|
||||
def fast_hash(data: str | bytes, length: int = 16) -> str:
|
||||
"""Fast non-cryptographic content hash for caches and dedup.
|
||||
|
||||
Uses MD5 (2-3x faster than SHA256). Not used for security — only for
|
||||
content-addressable lookups in compression caches, prefix tracking, etc.
|
||||
"""
|
||||
if isinstance(data, str):
|
||||
data = data.encode("utf-8")
|
||||
return hashlib.md5(data).hexdigest()[:length]
|
||||
|
||||
|
||||
def compute_messages_hash(messages: list[dict[str, Any]]) -> str:
|
||||
"""Compute hash of messages list for deduplication."""
|
||||
# Serialize deterministically
|
||||
|
|
@ -210,6 +222,9 @@ def format_cost(cost: float) -> str:
|
|||
|
||||
|
||||
def deep_copy_messages(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
"""Create a deep copy of messages list."""
|
||||
result: list[dict[str, Any]] = json.loads(json.dumps(messages))
|
||||
return result
|
||||
"""Create a deep copy of messages list.
|
||||
|
||||
Uses copy.deepcopy instead of json roundtrip (2-5x faster, avoids
|
||||
serialisation overhead on large conversation histories).
|
||||
"""
|
||||
return copy.deepcopy(messages)
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ build-backend = "hatchling.build"
|
|||
|
||||
[project]
|
||||
name = "headroom-ai"
|
||||
version = "0.5.5"
|
||||
version = "0.5.6"
|
||||
description = "The Context Optimization Layer for LLM Applications - Cut costs by 50-90%"
|
||||
readme = "README.md"
|
||||
license = "Apache-2.0"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue