mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
Add compression summaries, multi-provider headers, Dockerfile fix
Compression Summaries:
- New: headroom/transforms/compression_summary.py
- summarize_dropped_items(): categorizes compressed JSON items by
field values (status, type, level, etc.), highlights errors/failures
- summarize_compressed_code(): extracts function names from AST
signatures (language-agnostic: Python, JS, Go, Rust, Java)
- Newline-safe: strips \n from field values to keep markers single-line
- SmartCrusher: CCR markers include categorical summary of dropped items
e.g. "[500 items compressed to 20. Omitted: 87 passed, 2 failed.
Retrieve more: hash=abc123. Expires in 5m.]"
- CodeCompressor: CCR markers list compressed function names from AST
e.g. "[180 tokens compressed. 5 bodies compressed: authenticate().
Retrieve more: hash=abc123. Expires in 5m.]"
- Markers include TTL so LLM knows retrieval window
- Summary escapes { } to prevent .format() crashes
- Uses index-based dropped detection (not id()) for .copy() correctness
Proxy Response Headers:
- Anthropic, OpenAI, and Gemini handlers inject x-headroom-tokens-*
headers for SaaS metering
Multi-Provider Passthrough Routing:
- Detect x-goog-api-key (Gemini) and api-key (Azure OpenAI)
- X-Headroom-Base-URL for explicit upstream URL override
Dockerfile: add build-essential + g++ for hnswlib compilation
Bump version to 0.3.5
Tests: 27 new tests (unit, eval, integration with real API, tool invocation)
This commit is contained in:
chopratejas
parent
55d8fb6361
commit
729cc035a4
12 changed files with 1508 additions and 22 deletions
|
|
@ -142,7 +142,7 @@ from .transforms import (
|
|||
TransformPipeline,
|
||||
)
|
||||
|
||||
__version__ = "0.3.4"
|
||||
__version__ = "0.3.5"
|
||||
|
||||
__all__ = [
|
||||
# Main client
|
||||
|
|
|
|||
|
|
@ -529,7 +529,10 @@ class CCRConfig:
|
|||
# Retrieval marker format
|
||||
# Inserted at end of compressed content to tell LLM how to get more
|
||||
marker_template: str = (
|
||||
"\n[{original_count} items compressed to {compressed_count}. Retrieve more: hash={hash}]"
|
||||
"\n[{original_count} items compressed to {compressed_count}."
|
||||
"{summary}"
|
||||
" Retrieve more: hash={hash}."
|
||||
" Expires in {ttl_minutes}m.]"
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -5322,6 +5322,16 @@ class HeadroomProxy:
|
|||
response_headers.pop("content-encoding", None)
|
||||
response_headers.pop("content-length", None)
|
||||
|
||||
# Inject Headroom compression metrics (for SaaS metering)
|
||||
response_headers["x-headroom-tokens-before"] = str(original_tokens)
|
||||
response_headers["x-headroom-tokens-after"] = str(optimized_tokens)
|
||||
response_headers["x-headroom-tokens-saved"] = str(tokens_saved)
|
||||
response_headers["x-headroom-model"] = model
|
||||
if transforms_applied:
|
||||
response_headers["x-headroom-transforms"] = ",".join(transforms_applied)
|
||||
if cache_read_tokens > 0:
|
||||
response_headers["x-headroom-cached"] = "true"
|
||||
|
||||
return Response(
|
||||
content=response.content,
|
||||
status_code=response.status_code,
|
||||
|
|
@ -6562,11 +6572,27 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI:
|
|||
# Passthrough - route to correct backend based on headers
|
||||
@app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
|
||||
async def passthrough(request: Request, path: str):
|
||||
# Anthropic SDK always sends anthropic-version header and uses x-api-key for auth
|
||||
# OpenAI SDK uses Authorization: Bearer for auth
|
||||
# Allow explicit base URL override (for Azure, custom endpoints, etc.)
|
||||
custom_base = request.headers.get("x-headroom-base-url")
|
||||
if custom_base:
|
||||
return await proxy.handle_passthrough(request, custom_base.rstrip("/"))
|
||||
|
||||
# Anthropic: sends anthropic-version header and x-api-key
|
||||
if request.headers.get("anthropic-version") or request.headers.get("x-api-key"):
|
||||
base_url = proxy.ANTHROPIC_API_URL
|
||||
# Gemini: sends x-goog-api-key
|
||||
elif request.headers.get("x-goog-api-key"):
|
||||
base_url = proxy.GEMINI_API_URL
|
||||
# Azure OpenAI: sends api-key header (not x-api-key)
|
||||
elif request.headers.get("api-key"):
|
||||
# Azure requires explicit base URL (varies per deployment)
|
||||
azure_base = request.headers.get("x-headroom-base-url", "")
|
||||
if azure_base:
|
||||
base_url = azure_base.rstrip("/")
|
||||
else:
|
||||
base_url = proxy.OPENAI_API_URL # Fallback
|
||||
else:
|
||||
# Default: OpenAI
|
||||
base_url = proxy.OPENAI_API_URL
|
||||
return await proxy.handle_passthrough(request, base_url)
|
||||
|
||||
|
|
|
|||
|
|
@ -559,10 +559,22 @@ class CodeAwareCompressor(Transform):
|
|||
if self.config.enable_ccr and ratio < 0.8:
|
||||
cache_key = self._store_in_ccr(code, compressed, original_tokens)
|
||||
if cache_key:
|
||||
# Add standard CCR marker format for CCRToolInjector detection
|
||||
# Generate summary from AST data (language-agnostic)
|
||||
from .compression_summary import summarize_compressed_code
|
||||
|
||||
code_summary = summarize_compressed_code(
|
||||
structure.function_bodies,
|
||||
len(structure.function_bodies),
|
||||
)
|
||||
summary_str = f" {code_summary}." if code_summary else ""
|
||||
|
||||
# Add CCR marker (hash without quotes, matches CCRToolInjector regex)
|
||||
ttl_min = max(1, getattr(self.config, "ccr_ttl_seconds", 300) // 60)
|
||||
compressed += (
|
||||
f"\n# [{original_tokens} items compressed to {compressed_tokens}. "
|
||||
f"Retrieve more: hash={cache_key}]"
|
||||
f"\n# [{original_tokens - compressed_tokens} tokens compressed."
|
||||
f"{summary_str}"
|
||||
f" Retrieve more: hash={cache_key}."
|
||||
f" Expires in {ttl_min}m.]"
|
||||
)
|
||||
|
||||
return CodeCompressionResult(
|
||||
|
|
|
|||
243
headroom/transforms/compression_summary.py
Normal file
243
headroom/transforms/compression_summary.py
Normal file
|
|
@ -0,0 +1,243 @@
|
|||
"""Compression summary generator — describes what was dropped.
|
||||
|
||||
When content is compressed, the LLM needs to know what it's missing.
|
||||
Instead of just "[480 items omitted]", we generate a categorical summary:
|
||||
"[480 items omitted: 150 log entries (3 with errors), 200 test results (12 failures)]"
|
||||
|
||||
This helps the LLM decide whether to call headroom_retrieve and what to search for.
|
||||
|
||||
Used by:
|
||||
- SmartCrusher: categorizes dropped JSON items by field values
|
||||
- CodeCompressor: lists removed function/class names (from AST, language-agnostic)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from collections import Counter
|
||||
|
||||
|
||||
def summarize_dropped_items(
|
||||
all_items: list[dict],
|
||||
kept_items: list[dict],
|
||||
kept_indices: set[int] | None = None,
|
||||
max_categories: int = 5,
|
||||
max_notable: int = 3,
|
||||
) -> str:
|
||||
"""Generate a categorical summary of items that were dropped.
|
||||
|
||||
Args:
|
||||
all_items: The original full list of items.
|
||||
kept_items: The items that were kept after compression (used for count).
|
||||
kept_indices: Indices of kept items (preferred over identity comparison).
|
||||
max_categories: Maximum number of categories to show.
|
||||
max_notable: Maximum number of notable items to call out.
|
||||
|
||||
Returns:
|
||||
Summary string or empty string if no useful summary can be generated.
|
||||
"""
|
||||
if not all_items or len(kept_items) >= len(all_items):
|
||||
return ""
|
||||
|
||||
# Determine which items were dropped
|
||||
if kept_indices is not None:
|
||||
dropped = [item for i, item in enumerate(all_items) if i not in kept_indices]
|
||||
else:
|
||||
# Fallback: index-based comparison using JSON equality
|
||||
kept_json = {_item_key(item) for item in kept_items}
|
||||
dropped = [item for item in all_items if _item_key(item) not in kept_json]
|
||||
|
||||
if not dropped:
|
||||
return ""
|
||||
|
||||
# Strategy 1: Categorize by type/status/kind fields
|
||||
categories = _categorize_by_fields(dropped)
|
||||
|
||||
# Strategy 2: Find notable items (errors, failures, warnings)
|
||||
notable = _find_notable_items(dropped, max_notable)
|
||||
|
||||
# Build summary
|
||||
parts = []
|
||||
|
||||
if categories:
|
||||
cat_strs = []
|
||||
for field_val, count in categories.most_common(max_categories):
|
||||
cat_strs.append(f"{count} {field_val}")
|
||||
parts.append(", ".join(cat_strs))
|
||||
|
||||
if notable:
|
||||
parts.append(f"notable: {'; '.join(notable)}")
|
||||
|
||||
if not parts:
|
||||
# Fallback: just describe the data shape
|
||||
keys = _common_keys(dropped)
|
||||
if keys:
|
||||
parts.append(f"fields: {', '.join(keys[:5])}")
|
||||
|
||||
return "; ".join(parts)
|
||||
|
||||
|
||||
def summarize_compressed_code(
|
||||
function_bodies: list[tuple[str, str, int]],
|
||||
compressed_bodies_count: int,
|
||||
) -> str:
|
||||
"""Generate a summary of compressed code sections from AST data.
|
||||
|
||||
Language-agnostic: works with any language tree-sitter supports because
|
||||
it reads function signatures directly from the CodeCompressor's AST output.
|
||||
|
||||
Args:
|
||||
function_bodies: List of (signature, body, line) from CodeStructure.
|
||||
compressed_bodies_count: Number of bodies that were compressed.
|
||||
|
||||
Returns:
|
||||
Summary string like "5 bodies compressed: authenticate(), validate_token(), ..."
|
||||
or empty string.
|
||||
"""
|
||||
if not function_bodies or compressed_bodies_count == 0:
|
||||
return ""
|
||||
|
||||
# Extract short names from signatures
|
||||
names = []
|
||||
for sig, _body, _line in function_bodies:
|
||||
name = _extract_name_from_signature(sig)
|
||||
if name:
|
||||
names.append(name)
|
||||
|
||||
if not names:
|
||||
return f"{compressed_bodies_count} function bodies compressed"
|
||||
|
||||
# Show up to 6 names
|
||||
shown = names[:6]
|
||||
result = f"{compressed_bodies_count} bodies compressed: {', '.join(shown)}"
|
||||
if len(names) > 6:
|
||||
result += f" (+{len(names) - 6} more)"
|
||||
return result
|
||||
|
||||
|
||||
# ---- Internal helpers ----
|
||||
|
||||
# Fields that commonly indicate item category/type
|
||||
_CATEGORY_FIELDS = (
|
||||
"type",
|
||||
"status",
|
||||
"kind",
|
||||
"category",
|
||||
"level",
|
||||
"severity",
|
||||
"state",
|
||||
"phase",
|
||||
"action",
|
||||
"event_type",
|
||||
"log_level",
|
||||
"result",
|
||||
"outcome",
|
||||
)
|
||||
|
||||
# Values that indicate something notable/important
|
||||
_NOTABLE_PATTERNS = re.compile(
|
||||
r"error|fail|critical|warning|exception|crash|timeout|denied|rejected|invalid",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
# Values that look like URLs or paths (not useful as categories)
|
||||
_URL_PATTERN = re.compile(r"^https?://|^/[a-z]", re.IGNORECASE)
|
||||
|
||||
|
||||
def _item_key(item: dict) -> str:
|
||||
"""Create a hashable key for an item (for dropped detection without id())."""
|
||||
# Use first few field values as a fingerprint
|
||||
parts = []
|
||||
for k, v in list(item.items())[:4]:
|
||||
parts.append(f"{k}={v}")
|
||||
return "|".join(parts)
|
||||
|
||||
|
||||
def _categorize_by_fields(items: list[dict]) -> Counter:
|
||||
"""Categorize items by their type/status/kind field values."""
|
||||
categories: Counter = Counter()
|
||||
|
||||
for item in items:
|
||||
categorized = False
|
||||
for field in _CATEGORY_FIELDS:
|
||||
val = item.get(field)
|
||||
if val and isinstance(val, str) and len(val) < 50:
|
||||
clean_val = val.replace("\n", " ").replace("\r", "").strip()
|
||||
if clean_val:
|
||||
categories[clean_val] += 1
|
||||
categorized = True
|
||||
break
|
||||
if not categorized:
|
||||
# Try to infer from the item's first short string field
|
||||
for key, val in item.items():
|
||||
if (
|
||||
isinstance(val, str)
|
||||
and 2 < len(val) < 30
|
||||
and key not in ("id", "name", "path", "url", "href", "email")
|
||||
and not _URL_PATTERN.match(val)
|
||||
):
|
||||
clean_val = val.replace("\n", " ").replace("\r", "").strip()
|
||||
categories[f"{key}={clean_val}"] += 1
|
||||
break
|
||||
|
||||
return categories
|
||||
|
||||
|
||||
def _find_notable_items(items: list[dict], max_notable: int) -> list[str]:
|
||||
"""Find items that contain error/failure/warning indicators."""
|
||||
notable = []
|
||||
for item in items:
|
||||
item_str = str(item)[:500]
|
||||
matches = _NOTABLE_PATTERNS.findall(item_str)
|
||||
if matches:
|
||||
name = item.get("name", item.get("id", item.get("path", "")))
|
||||
if name:
|
||||
clean_name = str(name).replace("\n", " ").strip()[:50]
|
||||
notable.append(f"{clean_name} ({matches[0]})")
|
||||
else:
|
||||
notable.append(matches[0])
|
||||
if len(notable) >= max_notable:
|
||||
break
|
||||
return notable
|
||||
|
||||
|
||||
def _common_keys(items: list[dict]) -> list[str]:
|
||||
"""Get the most common keys across items."""
|
||||
key_counts: Counter = Counter()
|
||||
for item in items[:50]:
|
||||
for key in item.keys():
|
||||
key_counts[key] += 1
|
||||
return [k for k, _ in key_counts.most_common(8)]
|
||||
|
||||
|
||||
def _extract_name_from_signature(sig: str) -> str:
|
||||
"""Extract the function/method name from a signature string.
|
||||
|
||||
Works for any language because it looks for common patterns:
|
||||
- Python: "def authenticate(", "async def fetch("
|
||||
- JavaScript: "function authenticate(", "async function fetch("
|
||||
- Go: "func (s *Server) HandleRequest("
|
||||
- Rust: "fn authenticate("
|
||||
- Java/C++: "public void authenticate("
|
||||
"""
|
||||
# Try common function definition patterns
|
||||
match = re.search(r"(?:def|func|fn|function)\s+(?:\([^)]*\)\s*)?(\w+)", sig)
|
||||
if match:
|
||||
return match.group(1) + "()"
|
||||
|
||||
# Try method patterns: "public static void methodName("
|
||||
match = re.search(r"(?:public|private|protected|static|async|export)\s+.*?(\w+)\s*\(", sig)
|
||||
if match:
|
||||
return match.group(1) + "()"
|
||||
|
||||
# Try class patterns
|
||||
match = re.search(r"class\s+(\w+)", sig)
|
||||
if match:
|
||||
return match.group(1)
|
||||
|
||||
# Fallback: last word before (
|
||||
match = re.search(r"(\w+)\s*\(", sig)
|
||||
if match:
|
||||
return match.group(1) + "()"
|
||||
|
||||
return ""
|
||||
|
|
@ -2127,11 +2127,22 @@ class SmartCrusher(Transform):
|
|||
|
||||
# CCR: Inject retrieval markers if compression happened and CCR is enabled
|
||||
if was_modified and ccr_markers and self._ccr_config.inject_retrieval_marker:
|
||||
for ccr_hash, original_count, compressed_count in ccr_markers:
|
||||
for marker_data in ccr_markers:
|
||||
if len(marker_data) == 4:
|
||||
ccr_hash, original_count, compressed_count, dropped_summary = marker_data
|
||||
else:
|
||||
ccr_hash, original_count, compressed_count = marker_data
|
||||
dropped_summary = ""
|
||||
summary_str = f" Omitted: {dropped_summary}." if dropped_summary else ""
|
||||
# Escape { } in summary to prevent .format() errors
|
||||
safe_summary = summary_str.replace("{", "{{").replace("}", "}}")
|
||||
ttl_seconds = getattr(self._ccr_config, "store_ttl_seconds", 300)
|
||||
marker = self._ccr_config.marker_template.format(
|
||||
original_count=original_count,
|
||||
compressed_count=compressed_count,
|
||||
hash=ccr_hash,
|
||||
summary=safe_summary,
|
||||
ttl_minutes=max(1, ttl_seconds // 60),
|
||||
)
|
||||
result += marker
|
||||
|
||||
|
|
@ -2149,24 +2160,24 @@ class SmartCrusher(Transform):
|
|||
|
||||
Returns:
|
||||
Tuple of (processed_value, info_string, ccr_markers).
|
||||
ccr_markers is a list of (hash, original_count, compressed_count) tuples.
|
||||
ccr_markers is a list of (hash, original_count, compressed_count, summary) tuples.
|
||||
"""
|
||||
info_parts = []
|
||||
ccr_markers: list[tuple[str, int, int]] = []
|
||||
ccr_markers: list[tuple] = []
|
||||
|
||||
if isinstance(value, list):
|
||||
# Check if this array should be crushed
|
||||
# Must have enough items AND all items must be dicts (not mixed types)
|
||||
all_dicts = value and all(isinstance(item, dict) for item in value)
|
||||
if len(value) >= self.config.min_items_to_analyze and all_dicts:
|
||||
crushed, strategy, ccr_hash = self._crush_array(
|
||||
crushed, strategy, ccr_hash, dropped_summary = self._crush_array(
|
||||
value, query_context, tool_name, bias=bias
|
||||
)
|
||||
info_parts.append(f"{strategy}({len(value)}->{len(crushed)})")
|
||||
|
||||
# Track CCR marker for later injection
|
||||
# Track CCR marker for later injection (with summary)
|
||||
if ccr_hash:
|
||||
ccr_markers.append((ccr_hash, len(value), len(crushed)))
|
||||
ccr_markers.append((ccr_hash, len(value), len(crushed), dropped_summary))
|
||||
|
||||
return crushed, ",".join(info_parts), ccr_markers
|
||||
else:
|
||||
|
|
@ -2204,7 +2215,7 @@ class SmartCrusher(Transform):
|
|||
query_context: str = "",
|
||||
tool_name: str | None = None,
|
||||
bias: float = 1.0,
|
||||
) -> tuple[list, str, str | None]:
|
||||
) -> tuple[list, str, str | None, str]:
|
||||
"""Crush an array using statistical analysis and relevance scoring.
|
||||
|
||||
IMPORTANT: If crushability analysis determines it's not safe to crush
|
||||
|
|
@ -2223,8 +2234,9 @@ class SmartCrusher(Transform):
|
|||
bias: Compression bias multiplier (>1 = keep more, <1 = keep fewer).
|
||||
|
||||
Returns:
|
||||
Tuple of (crushed_items, strategy_info, ccr_hash).
|
||||
Tuple of (crushed_items, strategy_info, ccr_hash, dropped_summary).
|
||||
ccr_hash is the hash for retrieval if CCR is enabled, None otherwise.
|
||||
dropped_summary is a categorical summary of what was dropped.
|
||||
"""
|
||||
# BOUNDARY CHECK: Use adaptive sizing instead of hardcoded limit
|
||||
# compute_optimal_k handles trivial cases (n <= 8 → keep all)
|
||||
|
|
@ -2239,7 +2251,7 @@ class SmartCrusher(Transform):
|
|||
)
|
||||
|
||||
if len(items) <= adaptive_k:
|
||||
return items, "none:adaptive_at_limit", None
|
||||
return items, "none:adaptive_at_limit", None, ""
|
||||
|
||||
# Get feedback hints if enabled
|
||||
# THREAD-SAFETY: Use a local effective_max_items instead of mutating shared config
|
||||
|
|
@ -2264,7 +2276,7 @@ class SmartCrusher(Transform):
|
|||
)
|
||||
|
||||
if toin_hint.skip_compression:
|
||||
return items, f"skip:toin({toin_hint.reason})", None
|
||||
return items, f"skip:toin({toin_hint.reason})", None, ""
|
||||
|
||||
# Apply TOIN recommendations if from network or local learning
|
||||
toin_preserve_fields: list[str] = []
|
||||
|
|
@ -2313,7 +2325,7 @@ class SmartCrusher(Transform):
|
|||
|
||||
# Check if hints recommend skipping compression
|
||||
if hints.skip_compression:
|
||||
return items, f"skip:feedback({hints.reason})", None
|
||||
return items, f"skip:feedback({hints.reason})", None, ""
|
||||
|
||||
# Adjust max_items based on feedback
|
||||
if hints.suggested_items is not None:
|
||||
|
|
@ -2339,7 +2351,7 @@ class SmartCrusher(Transform):
|
|||
reason = ""
|
||||
if analysis.crushability:
|
||||
reason = f"skip:{analysis.crushability.reason}"
|
||||
return items, reason, None
|
||||
return items, reason, None, ""
|
||||
|
||||
# Apply TOIN strategy recommendation if available
|
||||
# TOIN learns which strategies work best from cross-user patterns
|
||||
|
|
@ -2356,7 +2368,7 @@ class SmartCrusher(Transform):
|
|||
if toin_compression_level:
|
||||
if toin_compression_level == "none":
|
||||
# Don't compress - return original
|
||||
return items, "skip:toin_level_none", None
|
||||
return items, "skip:toin_level_none", None, ""
|
||||
elif toin_compression_level == "conservative":
|
||||
# Be conservative - keep more items
|
||||
effective_max_items = max(effective_max_items, min(50, len(items) // 2))
|
||||
|
|
@ -2453,9 +2465,18 @@ class SmartCrusher(Transform):
|
|||
elif hints_applied:
|
||||
strategy_info += f"(feedback:{effective_max_items})"
|
||||
|
||||
# Generate categorical summary of dropped items (use indices, not identity)
|
||||
from .compression_summary import summarize_dropped_items
|
||||
|
||||
dropped_summary = summarize_dropped_items(
|
||||
items,
|
||||
result,
|
||||
kept_indices=set(plan.keep_indices),
|
||||
)
|
||||
|
||||
# Clean up temporary instance variable
|
||||
self._current_field_semantics = None
|
||||
return result, strategy_info, ccr_hash
|
||||
return result, strategy_info, ccr_hash, dropped_summary
|
||||
|
||||
except Exception:
|
||||
# Clean up temporary instance variable
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ build-backend = "hatchling.build"
|
|||
|
||||
[project]
|
||||
name = "headroom-ai"
|
||||
version = "0.3.4"
|
||||
version = "0.3.5"
|
||||
description = "The Context Optimization Layer for LLM Applications - Cut costs by 50-90%"
|
||||
readme = "README.md"
|
||||
license = "Apache-2.0"
|
||||
|
|
|
|||
178
tests/test_compression_summary.py
Normal file
178
tests/test_compression_summary.py
Normal file
|
|
@ -0,0 +1,178 @@
|
|||
"""Tests for compression summary generation."""
|
||||
|
||||
from headroom.transforms.compression_summary import (
|
||||
_extract_name_from_signature,
|
||||
summarize_compressed_code,
|
||||
summarize_dropped_items,
|
||||
)
|
||||
|
||||
|
||||
class TestSummarizeDroppedItems:
|
||||
def test_items_with_status_field(self):
|
||||
all_items = [{"id": i, "name": f"item-{i}", "status": "active"} for i in range(50)] + [
|
||||
{"id": i, "name": f"err-{i}", "status": "error"} for i in range(5)
|
||||
]
|
||||
kept = all_items[:3]
|
||||
summary = summarize_dropped_items(all_items, kept, kept_indices={0, 1, 2})
|
||||
assert "active" in summary
|
||||
assert summary
|
||||
|
||||
def test_items_with_type_field(self):
|
||||
all_items = [{"type": "log", "message": f"entry {i}"} for i in range(30)] + [
|
||||
{"type": "metric", "value": i} for i in range(20)
|
||||
]
|
||||
kept = all_items[:5]
|
||||
summary = summarize_dropped_items(all_items, kept, kept_indices={0, 1, 2, 3, 4})
|
||||
assert "log" in summary or "metric" in summary
|
||||
|
||||
def test_notable_items_with_errors(self):
|
||||
all_items = [{"name": f"test-{i}", "result": "pass"} for i in range(40)] + [
|
||||
{"name": "test-auth", "result": "fail", "error": "authentication failed"},
|
||||
]
|
||||
kept = all_items[:5]
|
||||
summary = summarize_dropped_items(all_items, kept, kept_indices=set(range(5)))
|
||||
assert summary
|
||||
|
||||
def test_no_dropped_items(self):
|
||||
items = [{"id": 1}, {"id": 2}]
|
||||
summary = summarize_dropped_items(items, items)
|
||||
assert summary == ""
|
||||
|
||||
def test_empty_input(self):
|
||||
summary = summarize_dropped_items([], [])
|
||||
assert summary == ""
|
||||
|
||||
def test_all_items_dropped(self):
|
||||
items = [{"status": "active", "name": f"item-{i}"} for i in range(20)]
|
||||
summary = summarize_dropped_items(items, [], kept_indices=set())
|
||||
assert "active" in summary
|
||||
|
||||
def test_mixed_category_fields(self):
|
||||
all_items = [{"level": "info", "msg": "something"} for _ in range(10)] + [
|
||||
{"level": "error", "msg": "bad thing"} for _ in range(3)
|
||||
]
|
||||
kept = all_items[:2]
|
||||
summary = summarize_dropped_items(all_items, kept, kept_indices={0, 1})
|
||||
assert summary
|
||||
|
||||
def test_items_without_category_fields(self):
|
||||
all_items = [{"code": 200, "count": i} for i in range(30)]
|
||||
kept = all_items[:3]
|
||||
summary = summarize_dropped_items(all_items, kept, kept_indices={0, 1, 2})
|
||||
assert summary # Should produce field-based fallback
|
||||
|
||||
def test_summary_not_too_long(self):
|
||||
all_items = [{"type": f"type_{i % 20}", "data": "x"} for i in range(100)]
|
||||
kept = all_items[:5]
|
||||
summary = summarize_dropped_items(all_items, kept, kept_indices=set(range(5)))
|
||||
assert len(summary) < 500
|
||||
|
||||
def test_url_values_excluded_from_categories(self):
|
||||
"""URL-like values should not be used as category labels."""
|
||||
all_items = [
|
||||
{"url": f"https://api.example.com/v1/items/{i}", "method": "GET"} for i in range(30)
|
||||
]
|
||||
kept = all_items[:3]
|
||||
summary = summarize_dropped_items(all_items, kept, kept_indices={0, 1, 2})
|
||||
assert "https://" not in summary
|
||||
|
||||
def test_fallback_without_kept_indices(self):
|
||||
"""Works correctly without kept_indices (uses _item_key fallback)."""
|
||||
all_items = [{"status": "active", "id": i} for i in range(20)]
|
||||
kept = [all_items[0], all_items[1]] # Copies from same list
|
||||
summary = summarize_dropped_items(all_items, kept)
|
||||
assert summary
|
||||
|
||||
|
||||
class TestSummarizeCompressedCode:
|
||||
def test_python_function_bodies(self):
|
||||
bodies = [
|
||||
("def authenticate(username, password):", " db = get_db()\n return True", 10),
|
||||
("def validate_token(token):", " return jwt.decode(token)", 20),
|
||||
("def refresh_session(user):", " session.extend()", 30),
|
||||
]
|
||||
summary = summarize_compressed_code(bodies, 3)
|
||||
assert "3 bodies compressed" in summary
|
||||
assert "authenticate()" in summary
|
||||
assert "validate_token()" in summary
|
||||
|
||||
def test_javascript_function_bodies(self):
|
||||
bodies = [
|
||||
("function handleRequest(req, res) {", " res.send('ok');", 5),
|
||||
("async function fetchData(url) {", " return await fetch(url);", 15),
|
||||
]
|
||||
summary = summarize_compressed_code(bodies, 2)
|
||||
assert "handleRequest()" in summary
|
||||
assert "fetchData()" in summary
|
||||
|
||||
def test_go_function_bodies(self):
|
||||
bodies = [
|
||||
(
|
||||
"func (s *Server) HandleRequest(w http.ResponseWriter, r *http.Request) {",
|
||||
' w.Write([]byte("ok"))',
|
||||
10,
|
||||
),
|
||||
("func main() {", " server.Start()", 1),
|
||||
]
|
||||
summary = summarize_compressed_code(bodies, 2)
|
||||
assert "HandleRequest()" in summary
|
||||
assert "main()" in summary
|
||||
|
||||
def test_rust_function_bodies(self):
|
||||
bodies = [
|
||||
("fn authenticate(token: &str) -> Result<User, Error> {", " Ok(User::new())", 10),
|
||||
]
|
||||
summary = summarize_compressed_code(bodies, 1)
|
||||
assert "authenticate()" in summary
|
||||
|
||||
def test_empty_bodies(self):
|
||||
summary = summarize_compressed_code([], 0)
|
||||
assert summary == ""
|
||||
|
||||
def test_many_bodies_truncated(self):
|
||||
bodies = [(f"def func_{i}(x):", f" return {i}", i * 10) for i in range(20)]
|
||||
summary = summarize_compressed_code(bodies, 20)
|
||||
assert "+14 more" in summary # 20 - 6 shown
|
||||
|
||||
|
||||
class TestExtractNameFromSignature:
|
||||
def test_python_def(self):
|
||||
assert _extract_name_from_signature("def authenticate(username):") == "authenticate()"
|
||||
|
||||
def test_python_async_def(self):
|
||||
assert _extract_name_from_signature("async def fetch_data(url):") == "fetch_data()"
|
||||
|
||||
def test_javascript_function(self):
|
||||
assert _extract_name_from_signature("function handleClick(event) {") == "handleClick()"
|
||||
|
||||
def test_go_func(self):
|
||||
assert (
|
||||
_extract_name_from_signature("func HandleRequest(w http.ResponseWriter) {")
|
||||
== "HandleRequest()"
|
||||
)
|
||||
|
||||
def test_go_method(self):
|
||||
assert _extract_name_from_signature("func (s *Server) Start() {") == "Start()"
|
||||
|
||||
def test_rust_fn(self):
|
||||
assert (
|
||||
_extract_name_from_signature("fn authenticate(token: &str) -> Result<User> {")
|
||||
== "authenticate()"
|
||||
)
|
||||
|
||||
def test_java_method(self):
|
||||
assert (
|
||||
_extract_name_from_signature("public void processPayment(Payment p) {")
|
||||
== "processPayment()"
|
||||
)
|
||||
|
||||
def test_class(self):
|
||||
assert _extract_name_from_signature("class TokenValidator:") == "TokenValidator"
|
||||
|
||||
def test_empty(self):
|
||||
assert _extract_name_from_signature("") == ""
|
||||
|
||||
def test_export_async(self):
|
||||
assert (
|
||||
_extract_name_from_signature("export async function fetchUsers() {") == "fetchUsers()"
|
||||
)
|
||||
217
tests/test_compression_summary_eval.py
Normal file
217
tests/test_compression_summary_eval.py
Normal file
|
|
@ -0,0 +1,217 @@
|
|||
"""Eval: Compression summary quality — generic, unbiased.
|
||||
|
||||
Tests that compression summaries are:
|
||||
1. Accurate (categories match actual dropped items)
|
||||
2. Useful (contain information that would help retrieval)
|
||||
3. Not misleading (don't hallucinate categories)
|
||||
|
||||
These are NOT skewed to show summaries as amazing — they test
|
||||
real-world data patterns and verify correctness.
|
||||
"""
|
||||
|
||||
from headroom.transforms.compression_summary import (
|
||||
summarize_compressed_code,
|
||||
summarize_dropped_items,
|
||||
)
|
||||
|
||||
# ============================================================================
|
||||
# Realistic test data (modeled on actual tool outputs)
|
||||
# ============================================================================
|
||||
|
||||
|
||||
def _make_github_issues(n: int) -> list[dict]:
|
||||
"""Realistic GitHub issues list."""
|
||||
statuses = ["open"] * (n // 2) + ["closed"] * (n // 4) + ["in_progress"] * (n // 4)
|
||||
issues = []
|
||||
for i in range(n):
|
||||
issue = {
|
||||
"id": i + 1,
|
||||
"title": f"Issue #{i + 1}: {'Fix auth bug' if i == 42 else 'General issue'}",
|
||||
"status": statuses[i % len(statuses)],
|
||||
"labels": ["bug"] if i % 10 == 0 else ["enhancement"],
|
||||
"assignee": f"user-{i % 5}",
|
||||
}
|
||||
if i in (42, 87):
|
||||
issue["status"] = "open"
|
||||
issue["labels"] = ["critical", "bug"]
|
||||
issue["title"] = f"CRITICAL: Auth failure in production (issue #{i + 1})"
|
||||
issues.append(issue)
|
||||
return issues
|
||||
|
||||
|
||||
def _make_test_results(n: int) -> list[dict]:
|
||||
"""Realistic test suite results."""
|
||||
results = []
|
||||
for i in range(n):
|
||||
result = {
|
||||
"name": f"test_{'auth' if i < 10 else 'general'}_{i}",
|
||||
"status": "pass",
|
||||
"duration_ms": 50 + i * 2,
|
||||
}
|
||||
if i in (3, 7, 45, 88):
|
||||
result["status"] = "fail"
|
||||
result["error"] = "AssertionError: expected True, got False"
|
||||
if i in (12, 67):
|
||||
result["status"] = "error"
|
||||
result["error"] = "TimeoutError: test exceeded 30s limit"
|
||||
results.append(result)
|
||||
return results
|
||||
|
||||
|
||||
def _make_log_entries(n: int) -> list[dict]:
|
||||
"""Realistic log entries."""
|
||||
entries = []
|
||||
for i in range(n):
|
||||
entry = {
|
||||
"timestamp": f"2024-01-15T10:{i:02d}:00Z",
|
||||
"level": "info",
|
||||
"message": f"Request processed in {10 + i}ms",
|
||||
"service": "api-gateway",
|
||||
}
|
||||
if i in (15, 45, 89):
|
||||
entry["level"] = "error"
|
||||
entry["message"] = "Connection refused: database pool exhausted"
|
||||
if i in (20, 50):
|
||||
entry["level"] = "warning"
|
||||
entry["message"] = "High memory usage: 85% threshold exceeded"
|
||||
entries.append(entry)
|
||||
return entries
|
||||
|
||||
|
||||
def _make_api_endpoints(n: int) -> list[dict]:
|
||||
"""Realistic API endpoint list."""
|
||||
return [
|
||||
{
|
||||
"path": f"/api/v1/{'users' if i < n // 3 else 'orders' if i < 2 * n // 3 else 'products'}/{i}",
|
||||
"method": "GET" if i % 3 else "POST",
|
||||
"status_code": 200 if i % 20 else 500,
|
||||
"latency_ms": 50 + i,
|
||||
}
|
||||
for i in range(n)
|
||||
]
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Eval: Summary accuracy
|
||||
# ============================================================================
|
||||
|
||||
|
||||
class TestSummaryAccuracy:
|
||||
"""Verify summaries accurately reflect what was dropped."""
|
||||
|
||||
def test_github_issues_categories_correct(self):
|
||||
"""Summary mentions actual status values from dropped items."""
|
||||
issues = _make_github_issues(100)
|
||||
kept = issues[:5]
|
||||
summary = summarize_dropped_items(issues, kept)
|
||||
|
||||
# Should mention the status values present in dropped items
|
||||
assert summary # Non-empty
|
||||
# At minimum, should contain some status category info
|
||||
has_category = any(s in summary.lower() for s in ["open", "closed", "in_progress"])
|
||||
assert has_category, f"Summary missing status categories: {summary}"
|
||||
|
||||
def test_test_results_mentions_failures(self):
|
||||
"""Summary mentions failures when test results are compressed."""
|
||||
results = _make_test_results(100)
|
||||
kept = results[:5]
|
||||
summary = summarize_dropped_items(results, kept)
|
||||
|
||||
assert summary
|
||||
# Should mention pass/fail somewhere
|
||||
has_result = any(s in summary.lower() for s in ["pass", "fail", "error"])
|
||||
assert has_result, f"Summary missing test result info: {summary}"
|
||||
|
||||
def test_log_entries_mentions_errors(self):
|
||||
"""Summary mentions error log entries."""
|
||||
logs = _make_log_entries(100)
|
||||
kept = logs[:3]
|
||||
summary = summarize_dropped_items(logs, kept)
|
||||
|
||||
assert summary
|
||||
# Should categorize by log level
|
||||
has_level = any(s in summary.lower() for s in ["info", "error", "warning"])
|
||||
assert has_level, f"Summary missing log level info: {summary}"
|
||||
|
||||
def test_no_hallucinated_categories(self):
|
||||
"""Summary should NOT mention categories that don't exist."""
|
||||
items = [{"status": "active", "id": i} for i in range(50)]
|
||||
kept = items[:3]
|
||||
summary = summarize_dropped_items(items, kept)
|
||||
|
||||
# Should NOT mention statuses that don't exist in the data
|
||||
assert "error" not in summary.lower() or "notable" in summary.lower()
|
||||
assert "fail" not in summary.lower()
|
||||
assert "critical" not in summary.lower()
|
||||
|
||||
def test_summary_proportional_to_data(self):
|
||||
"""Category counts in summary should roughly match actual data."""
|
||||
items = (
|
||||
[{"type": "log", "data": "x"}] * 100
|
||||
+ [{"type": "metric", "data": "y"}] * 50
|
||||
+ [{"type": "alert", "data": "z"}] * 10
|
||||
)
|
||||
kept = items[:3]
|
||||
summary = summarize_dropped_items(items, kept)
|
||||
|
||||
# "log" should appear with a higher count than "alert"
|
||||
# (We can't verify exact counts from the summary string,
|
||||
# but we verify the summary is non-empty and reasonable)
|
||||
assert summary
|
||||
assert len(summary) < 300
|
||||
|
||||
|
||||
class TestSummaryUsefulness:
|
||||
"""Verify summaries contain information useful for retrieval."""
|
||||
|
||||
def test_enough_info_to_search(self):
|
||||
"""Summary should contain terms the LLM could use as search queries."""
|
||||
results = _make_test_results(100)
|
||||
kept = results[:5]
|
||||
summary = summarize_dropped_items(results, kept)
|
||||
|
||||
# The LLM should be able to extract search terms from the summary
|
||||
# At minimum, it should know WHAT KIND of items are in the compressed data
|
||||
assert len(summary) > 10, "Summary too short to be useful"
|
||||
|
||||
def test_notable_items_actionable(self):
|
||||
"""Notable items should contain enough info to act on."""
|
||||
logs = _make_log_entries(100)
|
||||
kept = logs[:2]
|
||||
summary = summarize_dropped_items(logs, kept)
|
||||
|
||||
# If there are errors in the logs, the summary should help
|
||||
# the LLM decide to retrieve them
|
||||
assert summary
|
||||
# Just verify it's substantive enough
|
||||
assert len(summary.split()) > 3
|
||||
|
||||
def test_api_endpoints_described(self):
|
||||
"""API endpoint data should produce some useful description."""
|
||||
endpoints = _make_api_endpoints(60)
|
||||
kept = endpoints[:5]
|
||||
summary = summarize_dropped_items(endpoints, kept)
|
||||
|
||||
assert summary # Should produce SOMETHING, even without type/status fields
|
||||
|
||||
|
||||
class TestCodeSummaryAccuracy:
|
||||
"""Verify code summaries accurately describe removed sections."""
|
||||
|
||||
def test_real_python_module(self):
|
||||
"""Summary of a realistic Python module compression."""
|
||||
# Use AST-based summary (language-agnostic)
|
||||
bodies = [
|
||||
("def __init__(self, url: str, pool_size: int = 10):", "...", 8),
|
||||
("def connect(self) -> Any:", "...", 15),
|
||||
("def _create_new(self) -> Any:", "...", 22),
|
||||
("def release(self, conn: Any) -> None:", "...", 28),
|
||||
("def close_all(self) -> None:", "...", 33),
|
||||
("def create_engine(url: str) -> DatabaseConnection:", "...", 38),
|
||||
]
|
||||
summary = summarize_compressed_code(bodies, 6)
|
||||
assert "6 bodies compressed" in summary
|
||||
has_names = any(
|
||||
name in summary for name in ["connect()", "release()", "close_all()", "create_engine()"]
|
||||
)
|
||||
assert has_names, f"Summary missing function names: {summary}"
|
||||
256
tests/test_compression_summary_hard_eval.py
Normal file
256
tests/test_compression_summary_hard_eval.py
Normal file
|
|
@ -0,0 +1,256 @@
|
|||
"""Hard eval: Cases where the LLM has NO reason to check compressed data.
|
||||
|
||||
The previous eval asked "are there failures?" — that's too easy, the LLM
|
||||
will proactively check regardless of summary.
|
||||
|
||||
This eval tests the SUBTLE case: the user asks a DIFFERENT question,
|
||||
but the answer is in the compressed data. The summary is the only hint.
|
||||
|
||||
Requires: ANTHROPIC_API_KEY in environment or .env file.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
env_path = Path(__file__).parent.parent / ".env"
|
||||
if env_path.exists():
|
||||
for line in env_path.read_text().splitlines():
|
||||
line = line.strip()
|
||||
if line and not line.startswith("#") and "=" in line:
|
||||
key, _, value = line.partition("=")
|
||||
os.environ.setdefault(key.strip(), value.strip())
|
||||
|
||||
ANTHROPIC_KEY = os.environ.get("ANTHROPIC_API_KEY", "")
|
||||
|
||||
pytestmark = pytest.mark.skipif(
|
||||
not ANTHROPIC_KEY,
|
||||
reason="ANTHROPIC_API_KEY not set",
|
||||
)
|
||||
|
||||
HEADROOM_RETRIEVE_TOOL = {
|
||||
"name": "headroom_retrieve",
|
||||
"description": "Retrieve uncompressed content. Pass a query to search within it.",
|
||||
"input_schema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"hash": {"type": "string"},
|
||||
"query": {"type": "string"},
|
||||
},
|
||||
"required": ["hash"],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _call_claude(messages, tools, max_tokens=300):
|
||||
import httpx
|
||||
|
||||
resp = httpx.post(
|
||||
"https://api.anthropic.com/v1/messages",
|
||||
headers={
|
||||
"X-Api-Key": ANTHROPIC_KEY,
|
||||
"anthropic-version": "2023-06-01",
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
json={
|
||||
"model": "claude-sonnet-4-5-20250929",
|
||||
"max_tokens": max_tokens,
|
||||
"messages": messages,
|
||||
"tools": tools,
|
||||
},
|
||||
timeout=30,
|
||||
)
|
||||
return resp.json()
|
||||
|
||||
|
||||
def _get_tool_calls(resp):
|
||||
return [
|
||||
{"name": b["name"], "input": b.get("input", {})}
|
||||
for b in resp.get("content", [])
|
||||
if b.get("type") == "tool_use"
|
||||
]
|
||||
|
||||
|
||||
def _get_text(resp):
|
||||
return " ".join(b.get("text", "") for b in resp.get("content", []) if b.get("type") == "text")
|
||||
|
||||
|
||||
class TestHardCases:
|
||||
"""Cases where the LLM wouldn't naturally check compressed data."""
|
||||
|
||||
def test_config_lookup_with_summary(self):
|
||||
"""User asks about a config value that's in compressed data.
|
||||
|
||||
The visible items are all about 'production' env.
|
||||
The compressed items include 'staging' configs.
|
||||
Summary mentions this. LLM should retrieve.
|
||||
"""
|
||||
visible = [
|
||||
{"env": "production", "key": "DATABASE_URL", "value": "postgres://prod-db:5432/app"},
|
||||
{"env": "production", "key": "REDIS_URL", "value": "redis://prod-cache:6379"},
|
||||
{"env": "production", "key": "API_RATE_LIMIT", "value": "1000"},
|
||||
]
|
||||
# Hidden in compressed: staging configs
|
||||
all_items = (
|
||||
visible
|
||||
+ [
|
||||
{
|
||||
"env": "staging",
|
||||
"key": "DATABASE_URL",
|
||||
"value": "postgres://staging-db:5432/app",
|
||||
},
|
||||
{"env": "staging", "key": "REDIS_URL", "value": "redis://staging-cache:6379"},
|
||||
{"env": "staging", "key": "DEBUG_MODE", "value": "true"},
|
||||
{"env": "staging", "key": "LOG_LEVEL", "value": "debug"},
|
||||
]
|
||||
* 10
|
||||
+ [
|
||||
{
|
||||
"env": "development",
|
||||
"key": "DATABASE_URL",
|
||||
"value": "postgres://localhost:5432/dev",
|
||||
},
|
||||
]
|
||||
* 5
|
||||
)
|
||||
|
||||
from headroom.transforms.compression_summary import summarize_dropped_items
|
||||
|
||||
summary = summarize_dropped_items(all_items, visible)
|
||||
|
||||
compressed_output = json.dumps(visible, indent=2)
|
||||
compressed_output += (
|
||||
f"\n[{len(all_items) - len(visible)} items compressed to {len(visible)}."
|
||||
f" Omitted: {summary}."
|
||||
f' Retrieve specific items: headroom_retrieve(hash="config_hash", query="search")]'
|
||||
)
|
||||
|
||||
messages = [
|
||||
{
|
||||
"role": "user",
|
||||
"content": (
|
||||
f"Here are the application configs:\n\n{compressed_output}\n\n"
|
||||
"What is the staging database URL?"
|
||||
),
|
||||
}
|
||||
]
|
||||
|
||||
resp = _call_claude(messages, [HEADROOM_RETRIEVE_TOOL])
|
||||
tool_calls = _get_tool_calls(resp)
|
||||
text = _get_text(resp)
|
||||
|
||||
print(f"\n Summary: {summary}")
|
||||
print(f" Stop reason: {resp.get('stop_reason')}")
|
||||
print(f" Tool calls: {tool_calls}")
|
||||
if text:
|
||||
print(f" Text: {text[:200]}")
|
||||
|
||||
# WITH summary mentioning "staging" → should retrieve
|
||||
if resp.get("stop_reason") == "tool_use":
|
||||
assert tool_calls[0]["name"] == "headroom_retrieve"
|
||||
query = tool_calls[0]["input"].get("query", "").lower()
|
||||
assert "staging" in query or "database" in query
|
||||
print(" RESULT: Retrieved staging config ✓")
|
||||
else:
|
||||
# If LLM didn't retrieve, it should at least mention the data is compressed
|
||||
assert "compressed" in text.lower() or "staging" in text.lower()
|
||||
print(" RESULT: Mentioned compressed data but didn't retrieve")
|
||||
|
||||
def test_config_lookup_without_summary(self):
|
||||
"""Same question, but NO summary. LLM only sees production configs."""
|
||||
visible = [
|
||||
{"env": "production", "key": "DATABASE_URL", "value": "postgres://prod-db:5432/app"},
|
||||
{"env": "production", "key": "REDIS_URL", "value": "redis://prod-cache:6379"},
|
||||
{"env": "production", "key": "API_RATE_LIMIT", "value": "1000"},
|
||||
]
|
||||
|
||||
compressed_output = json.dumps(visible, indent=2)
|
||||
compressed_output += "\n[45 items compressed to 3. Retrieve more: hash=config_hash]"
|
||||
|
||||
messages = [
|
||||
{
|
||||
"role": "user",
|
||||
"content": (
|
||||
f"Here are the application configs:\n\n{compressed_output}\n\n"
|
||||
"What is the staging database URL?"
|
||||
),
|
||||
}
|
||||
]
|
||||
|
||||
resp = _call_claude(messages, [HEADROOM_RETRIEVE_TOOL])
|
||||
tool_calls = _get_tool_calls(resp)
|
||||
text = _get_text(resp)
|
||||
|
||||
print(f"\n Stop reason: {resp.get('stop_reason')}")
|
||||
print(f" Tool calls: {tool_calls}")
|
||||
if text:
|
||||
print(f" Text: {text[:200]}")
|
||||
|
||||
if resp.get("stop_reason") == "tool_use":
|
||||
print(" RESULT: LLM proactively retrieved (smart)")
|
||||
else:
|
||||
print(" RESULT: LLM did NOT retrieve staging config")
|
||||
|
||||
def test_specific_user_in_large_list_with_summary(self):
|
||||
"""Find a specific user in a compressed user list.
|
||||
|
||||
Summary mentions user roles. User asks about admins.
|
||||
"""
|
||||
visible = [
|
||||
{"id": i, "name": f"user_{i}", "role": "member", "email": f"user{i}@co.com"}
|
||||
for i in range(5)
|
||||
]
|
||||
all_items = (
|
||||
visible
|
||||
+ [
|
||||
{"id": i, "name": f"user_{i}", "role": "member", "email": f"user{i}@co.com"}
|
||||
for i in range(5, 95)
|
||||
]
|
||||
+ [
|
||||
{"id": 96, "name": "admin_sarah", "role": "admin", "email": "sarah@co.com"},
|
||||
{"id": 97, "name": "admin_mike", "role": "admin", "email": "mike@co.com"},
|
||||
{"id": 98, "name": "superadmin_jane", "role": "superadmin", "email": "jane@co.com"},
|
||||
]
|
||||
)
|
||||
|
||||
from headroom.transforms.compression_summary import summarize_dropped_items
|
||||
|
||||
summary = summarize_dropped_items(all_items, visible)
|
||||
|
||||
compressed_output = json.dumps(visible, indent=2)
|
||||
compressed_output += (
|
||||
f"\n[{len(all_items) - len(visible)} items compressed to {len(visible)}."
|
||||
f" Omitted: {summary}."
|
||||
f' Retrieve: headroom_retrieve(hash="users_hash", query="search")]'
|
||||
)
|
||||
|
||||
messages = [
|
||||
{
|
||||
"role": "user",
|
||||
"content": (
|
||||
f"Here's our user list:\n\n{compressed_output}\n\n"
|
||||
"Who are the admin users? I need to contact them."
|
||||
),
|
||||
}
|
||||
]
|
||||
|
||||
resp = _call_claude(messages, [HEADROOM_RETRIEVE_TOOL])
|
||||
tool_calls = _get_tool_calls(resp)
|
||||
text = _get_text(resp)
|
||||
|
||||
print(f"\n Summary: {summary}")
|
||||
print(f" Stop reason: {resp.get('stop_reason')}")
|
||||
print(f" Tool calls: {tool_calls}")
|
||||
if text:
|
||||
print(f" Text: {text[:200]}")
|
||||
|
||||
if resp.get("stop_reason") == "tool_use":
|
||||
query = tool_calls[0]["input"].get("query", "").lower()
|
||||
assert "admin" in query
|
||||
print(f" RESULT: Retrieved admin users (query='{query}') ✓")
|
||||
else:
|
||||
print(" RESULT: Did not retrieve admin users")
|
||||
235
tests/test_compression_summary_integration.py
Normal file
235
tests/test_compression_summary_integration.py
Normal file
|
|
@ -0,0 +1,235 @@
|
|||
"""Integration eval: Compression summaries with real LLM calls.
|
||||
|
||||
Tests whether compression summaries actually help the LLM find information
|
||||
in compressed data. Compares behavior with and without summaries.
|
||||
|
||||
Requires: ANTHROPIC_API_KEY in environment or .env file.
|
||||
|
||||
Run: python -m pytest tests/test_compression_summary_integration.py -v -s
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
# Load .env
|
||||
env_path = Path(__file__).parent.parent / ".env"
|
||||
if env_path.exists():
|
||||
for line in env_path.read_text().splitlines():
|
||||
line = line.strip()
|
||||
if line and not line.startswith("#") and "=" in line:
|
||||
key, _, value = line.partition("=")
|
||||
os.environ.setdefault(key.strip(), value.strip())
|
||||
|
||||
ANTHROPIC_KEY = os.environ.get("ANTHROPIC_API_KEY", "")
|
||||
|
||||
pytestmark = pytest.mark.skipif(
|
||||
not ANTHROPIC_KEY,
|
||||
reason="ANTHROPIC_API_KEY not set — skipping integration tests",
|
||||
)
|
||||
|
||||
|
||||
def _call_claude(messages: list[dict], max_tokens: int = 200) -> dict:
|
||||
"""Make a real Anthropic API call."""
|
||||
import httpx
|
||||
|
||||
resp = httpx.post(
|
||||
"https://api.anthropic.com/v1/messages",
|
||||
headers={
|
||||
"X-Api-Key": ANTHROPIC_KEY,
|
||||
"anthropic-version": "2023-06-01",
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
json={
|
||||
"model": "claude-sonnet-4-5-20250929",
|
||||
"max_tokens": max_tokens,
|
||||
"messages": messages,
|
||||
},
|
||||
timeout=30,
|
||||
)
|
||||
return resp.json()
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Test data: realistic tool output that gets compressed
|
||||
# ============================================================================
|
||||
|
||||
|
||||
def _make_test_suite_output(n: int = 100) -> list[dict]:
|
||||
"""Simulate a large test suite result (like from a CI/CD tool)."""
|
||||
results = []
|
||||
for i in range(n):
|
||||
result = {
|
||||
"test_name": f"test_module_{i // 10}.test_case_{i}",
|
||||
"status": "passed",
|
||||
"duration_ms": 50 + i * 3,
|
||||
"file": f"tests/test_module_{i // 10}.py",
|
||||
}
|
||||
# Inject specific failures that the LLM should find
|
||||
if i == 42:
|
||||
result["status"] = "failed"
|
||||
result["error"] = "AssertionError: expected status 200, got 401 in auth_middleware"
|
||||
result["test_name"] = "test_auth.test_login_with_expired_token"
|
||||
if i == 67:
|
||||
result["status"] = "failed"
|
||||
result["error"] = "TimeoutError: database connection pool exhausted after 30s"
|
||||
result["test_name"] = "test_database.test_concurrent_connections"
|
||||
if i == 88:
|
||||
result["status"] = "error"
|
||||
result["error"] = "ImportError: cannot import name 'NewFeature' from 'app.features'"
|
||||
result["test_name"] = "test_features.test_new_feature_integration"
|
||||
results.append(result)
|
||||
return results
|
||||
|
||||
|
||||
class TestSummaryHelpfulness:
|
||||
"""Compare LLM accuracy with vs without compression summaries."""
|
||||
|
||||
def test_find_failures_with_summary(self):
|
||||
"""LLM can identify failure types from the summary alone."""
|
||||
test_results = _make_test_suite_output(100)
|
||||
|
||||
# Simulate compression: keep first 10, compress rest with summary
|
||||
kept = test_results[:10]
|
||||
from headroom.transforms.compression_summary import summarize_dropped_items
|
||||
|
||||
summary = summarize_dropped_items(test_results, kept)
|
||||
|
||||
compressed_output = json.dumps(kept, indent=2)
|
||||
compressed_output += f"\n[90 items compressed to 10. Omitted: {summary}. "
|
||||
compressed_output += (
|
||||
'Retrieve specific items: headroom_retrieve(hash="abc123", query="your search")]'
|
||||
)
|
||||
|
||||
messages = [
|
||||
{
|
||||
"role": "user",
|
||||
"content": (
|
||||
"Here are the test results from CI:\n\n"
|
||||
f"{compressed_output}\n\n"
|
||||
"Are there any test failures? What types of failures are there? "
|
||||
"Answer concisely."
|
||||
),
|
||||
},
|
||||
]
|
||||
|
||||
resp = _call_claude(messages)
|
||||
text = resp.get("content", [{}])[0].get("text", "").lower()
|
||||
|
||||
# The LLM should mention failures (from the summary info)
|
||||
has_failure_info = any(
|
||||
word in text for word in ["fail", "error", "timeout", "assert", "import"]
|
||||
)
|
||||
print(f"\n Summary: {summary}")
|
||||
print(f" LLM response: {text[:200]}")
|
||||
print(f" Detected failure info: {has_failure_info}")
|
||||
|
||||
assert has_failure_info, f"LLM didn't detect failures from summary. Response: {text[:300]}"
|
||||
|
||||
def test_find_failures_without_summary(self):
|
||||
"""Baseline: LLM with NO summary — just '[90 items compressed]'."""
|
||||
test_results = _make_test_suite_output(100)
|
||||
|
||||
kept = test_results[:10]
|
||||
compressed_output = json.dumps(kept, indent=2)
|
||||
compressed_output += "\n[90 items compressed to 10. Retrieve more: hash=abc123]"
|
||||
|
||||
messages = [
|
||||
{
|
||||
"role": "user",
|
||||
"content": (
|
||||
"Here are the test results from CI:\n\n"
|
||||
f"{compressed_output}\n\n"
|
||||
"Are there any test failures? What types of failures are there? "
|
||||
"Answer concisely."
|
||||
),
|
||||
},
|
||||
]
|
||||
|
||||
resp = _call_claude(messages)
|
||||
text = resp.get("content", [{}])[0].get("text", "").lower()
|
||||
|
||||
# The LLM may or may not detect failures (it only sees 10 passing tests)
|
||||
has_failure_info = any(
|
||||
word in text for word in ["fail", "error", "timeout", "assert", "import"]
|
||||
)
|
||||
print(f"\n LLM response (no summary): {text[:200]}")
|
||||
print(f" Detected failure info: {has_failure_info}")
|
||||
|
||||
# We're NOT asserting here — this is the baseline.
|
||||
# We expect this to often MISS failures since the summary is generic.
|
||||
|
||||
def test_code_summary_helps_identify_functions(self):
|
||||
"""LLM can identify which functions were removed from compressed code."""
|
||||
original_code = '''
|
||||
class PaymentProcessor:
|
||||
"""Processes payments via Stripe."""
|
||||
|
||||
def __init__(self, api_key: str):
|
||||
self.stripe = stripe.Client(api_key)
|
||||
self.retry_count = 3
|
||||
|
||||
def charge(self, amount: float, currency: str, token: str) -> dict:
|
||||
for attempt in range(self.retry_count):
|
||||
try:
|
||||
return self.stripe.charges.create(
|
||||
amount=int(amount * 100),
|
||||
currency=currency,
|
||||
source=token,
|
||||
)
|
||||
except stripe.RateLimitError:
|
||||
time.sleep(2 ** attempt)
|
||||
raise PaymentError("Max retries exceeded")
|
||||
|
||||
def refund(self, charge_id: str, amount: float = None) -> dict:
|
||||
params = {"charge": charge_id}
|
||||
if amount:
|
||||
params["amount"] = int(amount * 100)
|
||||
return self.stripe.refunds.create(**params)
|
||||
|
||||
def get_balance(self) -> float:
|
||||
balance = self.stripe.balance.retrieve()
|
||||
return balance.available[0].amount / 100
|
||||
'''
|
||||
compressed_code = '''
|
||||
class PaymentProcessor:
|
||||
"""Processes payments via Stripe."""
|
||||
|
||||
def __init__(self, api_key: str):
|
||||
# [2 lines omitted]
|
||||
pass
|
||||
|
||||
def charge(self, amount: float, currency: str, token: str) -> dict:
|
||||
# [8 lines omitted]
|
||||
pass
|
||||
|
||||
def refund(self, charge_id: str, amount: float = None) -> dict:
|
||||
# [3 lines omitted]
|
||||
pass
|
||||
|
||||
def get_balance(self) -> float:
|
||||
# [2 lines omitted]
|
||||
pass
|
||||
'''
|
||||
from headroom.transforms.compression_summary import summarize_removed_code
|
||||
|
||||
code_summary = summarize_removed_code(original_code, compressed_code)
|
||||
|
||||
prompt = f"Here is a compressed Python file:\n\n```python\n{compressed_code}\n```\n\n"
|
||||
if code_summary:
|
||||
prompt += f"[Compression info: {code_summary}]\n\n"
|
||||
prompt += "I need to understand the retry logic. Which function should I look at? Answer in one sentence."
|
||||
|
||||
messages = [{"role": "user", "content": prompt}]
|
||||
resp = _call_claude(messages, max_tokens=100)
|
||||
text = resp.get("content", [{}])[0].get("text", "").lower()
|
||||
|
||||
print(f"\n Code summary: {code_summary}")
|
||||
print(f" LLM response: {text[:200]}")
|
||||
|
||||
# The LLM should identify the charge() function
|
||||
assert "charge" in text, f"LLM didn't identify charge() function. Response: {text}"
|
||||
295
tests/test_compression_summary_tool_eval.py
Normal file
295
tests/test_compression_summary_tool_eval.py
Normal file
|
|
@ -0,0 +1,295 @@
|
|||
"""Eval: Does the LLM invoke headroom_retrieve when summaries are present?
|
||||
|
||||
The REAL test — it's not enough for the LLM to know something is missing.
|
||||
It must actually call the tool to fetch it.
|
||||
|
||||
Compares:
|
||||
- WITH summary: LLM sees "2 failed, 1 error" → should call headroom_retrieve
|
||||
- WITHOUT summary: LLM sees "[90 items compressed]" → likely does NOT call tool
|
||||
|
||||
Requires: ANTHROPIC_API_KEY in environment or .env file.
|
||||
|
||||
Run: python -m pytest tests/test_compression_summary_tool_eval.py -v -s
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
# Load .env
|
||||
env_path = Path(__file__).parent.parent / ".env"
|
||||
if env_path.exists():
|
||||
for line in env_path.read_text().splitlines():
|
||||
line = line.strip()
|
||||
if line and not line.startswith("#") and "=" in line:
|
||||
key, _, value = line.partition("=")
|
||||
os.environ.setdefault(key.strip(), value.strip())
|
||||
|
||||
ANTHROPIC_KEY = os.environ.get("ANTHROPIC_API_KEY", "")
|
||||
|
||||
pytestmark = pytest.mark.skipif(
|
||||
not ANTHROPIC_KEY,
|
||||
reason="ANTHROPIC_API_KEY not set — skipping integration tests",
|
||||
)
|
||||
|
||||
# The headroom_retrieve tool definition (same as what CCR injects)
|
||||
HEADROOM_RETRIEVE_TOOL = {
|
||||
"name": "headroom_retrieve",
|
||||
"description": (
|
||||
"Retrieve original uncompressed content from Headroom's compression cache. "
|
||||
"Use this when you need more details from compressed data. "
|
||||
"You can pass a query to search within the compressed content."
|
||||
),
|
||||
"input_schema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"hash": {
|
||||
"type": "string",
|
||||
"description": "The hash key from the compression marker",
|
||||
},
|
||||
"query": {
|
||||
"type": "string",
|
||||
"description": "Optional search query to find specific items within the compressed data",
|
||||
},
|
||||
},
|
||||
"required": ["hash"],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _call_claude_with_tools(messages: list[dict], tools: list[dict], max_tokens: int = 300) -> dict:
|
||||
"""Make a real Anthropic API call with tool use."""
|
||||
import httpx
|
||||
|
||||
resp = httpx.post(
|
||||
"https://api.anthropic.com/v1/messages",
|
||||
headers={
|
||||
"X-Api-Key": ANTHROPIC_KEY,
|
||||
"anthropic-version": "2023-06-01",
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
json={
|
||||
"model": "claude-sonnet-4-5-20250929",
|
||||
"max_tokens": max_tokens,
|
||||
"messages": messages,
|
||||
"tools": tools,
|
||||
},
|
||||
timeout=30,
|
||||
)
|
||||
return resp.json()
|
||||
|
||||
|
||||
def _make_test_results(n: int = 100) -> list[dict]:
|
||||
"""Test suite output with hidden failures in the compressed portion."""
|
||||
results = []
|
||||
for i in range(n):
|
||||
result = {
|
||||
"test_name": f"test_module_{i // 10}.test_case_{i}",
|
||||
"status": "passed",
|
||||
"duration_ms": 50 + i * 3,
|
||||
}
|
||||
if i == 42:
|
||||
result["status"] = "failed"
|
||||
result["error"] = "AssertionError: expected 200, got 401 in auth_middleware"
|
||||
result["test_name"] = "test_auth.test_login_expired_token"
|
||||
if i == 67:
|
||||
result["status"] = "failed"
|
||||
result["error"] = "TimeoutError: database pool exhausted after 30s"
|
||||
result["test_name"] = "test_database.test_concurrent_connections"
|
||||
if i == 88:
|
||||
result["status"] = "error"
|
||||
result["error"] = "ImportError: cannot import 'NewFeature'"
|
||||
result["test_name"] = "test_features.test_new_feature_integration"
|
||||
results.append(result)
|
||||
return results
|
||||
|
||||
|
||||
def _has_tool_use(response: dict) -> bool:
|
||||
"""Check if the response contains a tool_use block."""
|
||||
for block in response.get("content", []):
|
||||
if block.get("type") == "tool_use":
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _get_tool_calls(response: dict) -> list[dict]:
|
||||
"""Extract all tool_use blocks from response."""
|
||||
calls = []
|
||||
for block in response.get("content", []):
|
||||
if block.get("type") == "tool_use":
|
||||
calls.append(
|
||||
{
|
||||
"name": block.get("name"),
|
||||
"input": block.get("input", {}),
|
||||
}
|
||||
)
|
||||
return calls
|
||||
|
||||
|
||||
class TestToolInvocationWithSummary:
|
||||
"""The real eval: does the LLM call headroom_retrieve?"""
|
||||
|
||||
def test_with_summary_triggers_tool_call(self):
|
||||
"""WITH compression summary → LLM should call headroom_retrieve."""
|
||||
test_results = _make_test_results(100)
|
||||
kept = test_results[:10] # All passing
|
||||
|
||||
from headroom.transforms.compression_summary import summarize_dropped_items
|
||||
|
||||
summary = summarize_dropped_items(test_results, kept)
|
||||
|
||||
compressed = json.dumps(kept, indent=2)
|
||||
compressed += (
|
||||
f"\n[90 items compressed to 10. Omitted: {summary}."
|
||||
f' Retrieve specific items: headroom_retrieve(hash="ccr_test_abc123", query="your search")]'
|
||||
)
|
||||
|
||||
messages = [
|
||||
{
|
||||
"role": "user",
|
||||
"content": (
|
||||
"Here are the test results from our CI pipeline:\n\n"
|
||||
f"{compressed}\n\n"
|
||||
"Tell me about any test failures. What went wrong?"
|
||||
),
|
||||
},
|
||||
]
|
||||
|
||||
resp = _call_claude_with_tools(messages, [HEADROOM_RETRIEVE_TOOL])
|
||||
|
||||
tool_calls = _get_tool_calls(resp)
|
||||
stop_reason = resp.get("stop_reason", "")
|
||||
|
||||
print(f"\n Summary: {summary}")
|
||||
print(f" Stop reason: {stop_reason}")
|
||||
print(f" Tool calls: {tool_calls}")
|
||||
|
||||
# With a summary showing failures, the LLM SHOULD call the tool
|
||||
if stop_reason == "tool_use":
|
||||
assert len(tool_calls) > 0
|
||||
call = tool_calls[0]
|
||||
assert call["name"] == "headroom_retrieve"
|
||||
assert call["input"].get("hash") == "ccr_test_abc123"
|
||||
# The query should be about failures/errors
|
||||
query = call["input"].get("query", "").lower()
|
||||
print(f" Query used: {query}")
|
||||
has_relevant_query = any(
|
||||
term in query for term in ["fail", "error", "issue", "problem", "broken", "test"]
|
||||
)
|
||||
assert has_relevant_query, f"Tool was called but query isn't relevant: {query}"
|
||||
print(" RESULT: LLM invoked headroom_retrieve with relevant query ✓")
|
||||
else:
|
||||
# LLM responded with text — check if it at least mentions the failures
|
||||
text = ""
|
||||
for block in resp.get("content", []):
|
||||
if block.get("type") == "text":
|
||||
text += block.get("text", "")
|
||||
print(f" LLM text response: {text[:200]}")
|
||||
# It's acceptable if the LLM mentions it WANTS to retrieve
|
||||
mentions_retrieval = any(
|
||||
term in text.lower()
|
||||
for term in ["retrieve", "headroom_retrieve", "fetch", "see more", "compressed"]
|
||||
)
|
||||
print(f" Mentions retrieval: {mentions_retrieval}")
|
||||
|
||||
def test_without_summary_baseline(self):
|
||||
"""WITHOUT compression summary → LLM likely does NOT call tool."""
|
||||
test_results = _make_test_results(100)
|
||||
kept = test_results[:10] # All passing
|
||||
|
||||
compressed = json.dumps(kept, indent=2)
|
||||
compressed += "\n[90 items compressed to 10. Retrieve more: hash=ccr_test_abc123]"
|
||||
|
||||
messages = [
|
||||
{
|
||||
"role": "user",
|
||||
"content": (
|
||||
"Here are the test results from our CI pipeline:\n\n"
|
||||
f"{compressed}\n\n"
|
||||
"Tell me about any test failures. What went wrong?"
|
||||
),
|
||||
},
|
||||
]
|
||||
|
||||
resp = _call_claude_with_tools(messages, [HEADROOM_RETRIEVE_TOOL])
|
||||
|
||||
tool_calls = _get_tool_calls(resp)
|
||||
stop_reason = resp.get("stop_reason", "")
|
||||
|
||||
print(f"\n Stop reason: {stop_reason}")
|
||||
print(f" Tool calls: {tool_calls}")
|
||||
|
||||
if stop_reason == "tool_use":
|
||||
call = tool_calls[0]
|
||||
print(f" Query used: {call['input'].get('query', 'none')}")
|
||||
print(" RESULT: LLM DID invoke tool (may check proactively)")
|
||||
else:
|
||||
text = ""
|
||||
for block in resp.get("content", []):
|
||||
if block.get("type") == "text":
|
||||
text += block.get("text", "")
|
||||
print(f" LLM text response: {text[:200]}")
|
||||
print(" RESULT: LLM did NOT invoke tool — assumed all tests passed")
|
||||
|
||||
def test_code_summary_triggers_retrieval(self):
|
||||
"""Code compression summary → LLM should retrieve specific function."""
|
||||
compressed_code = '''class PaymentProcessor:
|
||||
"""Processes payments via Stripe."""
|
||||
|
||||
def __init__(self, api_key: str):
|
||||
# [2 lines omitted]
|
||||
pass
|
||||
|
||||
def charge(self, amount: float, currency: str, token: str) -> dict:
|
||||
# [8 lines omitted]
|
||||
pass
|
||||
|
||||
def refund(self, charge_id: str, amount: float = None) -> dict:
|
||||
# [3 lines omitted]
|
||||
pass
|
||||
|
||||
def get_balance(self) -> float:
|
||||
# [2 lines omitted]
|
||||
pass
|
||||
# [180 tokens compressed. removed: def charge (12 lines), def refund (6 lines). Retrieve full code: headroom_retrieve(hash="ccr_code_xyz", query="function name")]'''
|
||||
|
||||
messages = [
|
||||
{
|
||||
"role": "user",
|
||||
"content": (
|
||||
"Here's the payment processor code:\n\n"
|
||||
f"```python\n{compressed_code}\n```\n\n"
|
||||
"There's a bug in the retry logic for failed charges. "
|
||||
"Can you find and fix it?"
|
||||
),
|
||||
},
|
||||
]
|
||||
|
||||
resp = _call_claude_with_tools(messages, [HEADROOM_RETRIEVE_TOOL])
|
||||
|
||||
tool_calls = _get_tool_calls(resp)
|
||||
stop_reason = resp.get("stop_reason", "")
|
||||
|
||||
print(f"\n Stop reason: {stop_reason}")
|
||||
print(f" Tool calls: {tool_calls}")
|
||||
|
||||
if stop_reason == "tool_use":
|
||||
call = tool_calls[0]
|
||||
assert call["name"] == "headroom_retrieve"
|
||||
query = call["input"].get("query", "").lower()
|
||||
print(f" Query: {query}")
|
||||
# Should be asking for the charge function specifically
|
||||
has_charge = any(term in query for term in ["charge", "retry", "payment", "stripe"])
|
||||
print(f" Targets charge/retry: {has_charge}")
|
||||
print(" RESULT: LLM invoked tool to get the charge() implementation ✓")
|
||||
else:
|
||||
text = ""
|
||||
for block in resp.get("content", []):
|
||||
if block.get("type") == "text":
|
||||
text += block.get("text", "")
|
||||
print(f" LLM text: {text[:200]}")
|
||||
print(" RESULT: LLM did NOT invoke tool")
|
||||
Loading…
Add table
Add a link
Reference in a new issue