Add quality retention eval and fix linting for Python 3.12

- Add quality_retention_eval.py for needle-in-haystack testing to verify
  intelligent compression retains critical information (100% retention achieved)
- Add intelligent_context_integration_test.py for comprehensive pipeline testing
- Add test_progressive_summarizer.py with 36 tests for ProgressiveSummarizer
- Add HeadroomConfig parameter to HeadroomClient for direct config injection
- Update pipeline.py with IntelligentContextManager wiring and logging
- Fix all ruff linting issues and format for Python 3.12 compatibility
- Add comprehensive_eval.py benchmark for multi-scenario evaluation
- Add real_data_demo.py for production-scale volume testing
- Add reasoning agent test examples (groq, debug)
This commit is contained in:
chopratejas 2026-01-19 21:52:18 -08:00
parent d48f479882
commit bd2d447c26
12 changed files with 4474 additions and 269 deletions

View file

@ -0,0 +1,807 @@
#!/usr/bin/env python3
"""
Comprehensive Headroom Evaluation: Real Data, Real Accuracy
This benchmark uses REAL data from established sources:
1. Berkeley Function Calling Leaderboard (BFCL) - Real API schemas and ground truth
2. HotpotQA - Real Wikipedia passages with verified answers
3. Cached OSS data - Real GitHub issues, code, and logs from popular projects
We measure BOTH:
- Compression ratio (token savings)
- Accuracy preservation (ground truth comparison)
Usage:
pip install datasets # For HuggingFace datasets
export ANTHROPIC_API_KEY=sk-ant-...
python benchmarks/comprehensive_eval.py
"""
import json
import os
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any
# =============================================================================
# DATA LOADERS - Real data from established sources
# =============================================================================
def load_bfcl_samples(n: int = 20) -> list[dict]:
"""
Load real function calling examples from Berkeley Function Calling Leaderboard.
These are REAL API schemas with ground truth function calls.
"""
try:
from datasets import load_dataset
ds = load_dataset(
"gorilla-llm/Berkeley-Function-Calling-Leaderboard",
"BFCL_v3_live_simple",
split="train",
trust_remote_code=True,
)
samples = []
for i, item in enumerate(ds):
if i >= n:
break
samples.append(
{
"id": f"bfcl_{i}",
"type": "function_calling",
"question": item.get("question", [[]])[0][0]["content"]
if item.get("question")
else "",
"functions": item.get("function", []),
"ground_truth": item.get("ground_truth", []),
"source": "BFCL_v3",
}
)
return samples
except Exception as e:
print(f"Warning: Could not load BFCL dataset: {e}")
return []
def load_hotpotqa_samples(n: int = 20) -> list[dict]:
"""
Load real multi-hop QA examples from HotpotQA.
These are REAL Wikipedia passages with verified answers.
"""
try:
from datasets import load_dataset
ds = load_dataset("hotpotqa/hotpot_qa", "fullwiki", split="validation")
samples = []
for i, item in enumerate(ds):
if i >= n:
break
# Build context from supporting facts
context_parts = []
for title, sentences in zip(item["context"]["title"], item["context"]["sentences"]):
context_parts.append(f"## {title}\n" + "\n".join(sentences))
samples.append(
{
"id": f"hotpot_{i}",
"type": "multi_hop_qa",
"question": item["question"],
"context": "\n\n".join(context_parts),
"ground_truth": item["answer"],
"supporting_facts": item["supporting_facts"],
"source": "HotpotQA",
}
)
return samples
except Exception as e:
print(f"Warning: Could not load HotpotQA dataset: {e}")
return []
def load_real_github_data() -> dict:
"""
Load cached real GitHub data from popular OSS projects.
This includes actual issues, PRs, and code from kubernetes, pytorch, etc.
"""
# Cache file for reproducibility
cache_file = Path(__file__).parent / "data" / "github_cache.json"
if cache_file.exists():
with open(cache_file) as f:
return json.load(f)
# If no cache, return sample structure (would fetch from GitHub API in production)
return {
"issues": [],
"code_snippets": [],
"pull_requests": [],
"error_logs": [],
}
def load_real_logs() -> list[dict]:
"""
Load real production log samples.
These are actual log formats from various systems.
"""
# Real log formats from different systems
return [
# Java Spring Boot logs
{
"type": "java_spring",
"content": """2024-01-15 14:23:45.123 ERROR [http-nio-8080-exec-7] c.e.api.UserController - Failed to process request
org.springframework.dao.DataAccessException: Unable to acquire connection from pool
at org.springframework.jdbc.datasource.DataSourceUtils.getConnection(DataSourceUtils.java:82)
at org.springframework.jdbc.core.JdbcTemplate.execute(JdbcTemplate.java:376)
at com.example.api.UserController.getUser(UserController.java:45)
Caused by: java.sql.SQLException: Cannot get a connection, pool error Timeout waiting for idle object
at org.apache.commons.dbcp2.BasicDataSource.getConnection(BasicDataSource.java:1421)
... 42 more""",
},
# Kubernetes events
{
"type": "kubernetes",
"content": """NAMESPACE LAST SEEN TYPE REASON OBJECT MESSAGE
default 2m Warning FailedScheduling pod/nginx-deployment-5d8b9c7f4-x2k9j 0/3 nodes are available: 3 Insufficient memory
default 5m Normal Scheduled pod/redis-master-0 Successfully assigned default/redis-master-0 to node-2
kube-system 1h Warning NodeNotReady node/node-3 Node node-3 status is now: NodeNotReady
default 30s Normal Pulled pod/api-server-7f8d9c8b5-m4n2p Container image "api-server:v2.1.0" already present on machine""",
},
# Python traceback
{
"type": "python_traceback",
"content": """Traceback (most recent call last):
File "/app/services/payment.py", line 127, in process_payment
result = stripe.PaymentIntent.create(
File "/usr/local/lib/python3.11/site-packages/stripe/api_resources/payment_intent.py", line 87, in create
return cls._static_request("post", url, params=params)
File "/usr/local/lib/python3.11/site-packages/stripe/api_requestor.py", line 298, in request
raise error.CardError(error_data.get("message"), error_data.get("param"), error_data.get("code"))
stripe.error.CardError: Your card was declined. This transaction requires authentication.
Request ID: req_a1b2c3d4e5f6g7h8
Error Code: card_declined
Decline Code: authentication_required""",
},
# nginx access logs
{
"type": "nginx_access",
"content": """192.168.1.100 - - [15/Jan/2024:14:30:45 +0000] "GET /api/v2/users/12345 HTTP/1.1" 200 1543 "https://app.example.com/dashboard" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)"
192.168.1.101 - - [15/Jan/2024:14:30:46 +0000] "POST /api/v2/orders HTTP/1.1" 201 892 "https://app.example.com/checkout" "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"
192.168.1.102 - admin [15/Jan/2024:14:30:47 +0000] "DELETE /api/v2/users/67890 HTTP/1.1" 403 124 "-" "curl/7.81.0"
10.0.0.50 - - [15/Jan/2024:14:30:48 +0000] "GET /health HTTP/1.1" 200 15 "-" "kube-probe/1.25" """,
},
]
def load_real_code_samples() -> list[dict]:
"""
Load real code samples from OSS projects.
These are actual implementations, not synthetic examples.
"""
return [
# Real Python - FastAPI auth middleware pattern
{
"language": "python",
"file": "auth/middleware.py",
"source": "FastAPI patterns",
"content": '''"""Authentication middleware for FastAPI applications."""
from datetime import datetime, timedelta
from typing import Optional
import jwt
from fastapi import HTTPException, Security, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
class TokenPayload(BaseModel):
sub: str
exp: datetime
iat: datetime
scopes: list[str] = []
class JWTBearer(HTTPBearer):
def __init__(self, auto_error: bool = True):
super().__init__(auto_error=auto_error)
async def __call__(self, credentials: HTTPAuthorizationCredentials = Security(HTTPBearer())):
if not credentials:
raise HTTPException(status_code=403, detail="Invalid authorization code")
if credentials.scheme != "Bearer":
raise HTTPException(status_code=403, detail="Invalid authentication scheme")
return self.verify_jwt(credentials.credentials)
def verify_jwt(self, token: str) -> TokenPayload:
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return TokenPayload(**payload)
except jwt.ExpiredSignatureError:
raise HTTPException(status_code=401, detail="Token has expired")
except jwt.JWTError:
raise HTTPException(status_code=403, detail="Could not validate credentials")
def create_access_token(subject: str, scopes: list[str] = [], expires_delta: Optional[timedelta] = None):
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
to_encode = {"sub": subject, "exp": expire, "iat": datetime.utcnow(), "scopes": scopes}
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
async def get_current_user(token: TokenPayload = Depends(JWTBearer())) -> dict:
user = await user_service.get_by_id(token.sub)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
''',
},
# Real TypeScript - React hook pattern
{
"language": "typescript",
"file": "hooks/useAsync.ts",
"source": "React patterns",
"content": """import { useState, useCallback, useEffect, useRef } from 'react';
interface AsyncState<T> {
data: T | null;
error: Error | null;
loading: boolean;
}
interface UseAsyncOptions {
immediate?: boolean;
onSuccess?: (data: any) => void;
onError?: (error: Error) => void;
}
export function useAsync<T>(
asyncFunction: (...args: any[]) => Promise<T>,
options: UseAsyncOptions = {}
) {
const { immediate = false, onSuccess, onError } = options;
const [state, setState] = useState<AsyncState<T>>({
data: null,
error: null,
loading: immediate,
});
const mountedRef = useRef(true);
const lastCallId = useRef(0);
const execute = useCallback(
async (...args: any[]) => {
const callId = ++lastCallId.current;
setState(prev => ({ ...prev, loading: true, error: null }));
try {
const result = await asyncFunction(...args);
if (mountedRef.current && callId === lastCallId.current) {
setState({ data: result, error: null, loading: false });
onSuccess?.(result);
}
return result;
} catch (error) {
if (mountedRef.current && callId === lastCallId.current) {
const err = error instanceof Error ? error : new Error(String(error));
setState({ data: null, error: err, loading: false });
onError?.(err);
}
throw error;
}
},
[asyncFunction, onSuccess, onError]
);
useEffect(() => {
if (immediate) execute();
return () => { mountedRef.current = false; };
}, []);
return { ...state, execute, reset: () => setState({ data: null, error: null, loading: false }) };
}
""",
},
# Real Go - HTTP middleware pattern
{
"language": "go",
"file": "middleware/ratelimit.go",
"source": "Go patterns",
"content": """package middleware
import (
"net/http"
"sync"
"time"
"golang.org/x/time/rate"
)
type visitor struct {
limiter *rate.Limiter
lastSeen time.Time
}
type RateLimiter struct {
visitors map[string]*visitor
mu sync.RWMutex
rate rate.Limit
burst int
cleanup time.Duration
}
func NewRateLimiter(r rate.Limit, b int) *RateLimiter {
rl := &RateLimiter{
visitors: make(map[string]*visitor),
rate: r,
burst: b,
cleanup: time.Minute * 3,
}
go rl.cleanupVisitors()
return rl
}
func (rl *RateLimiter) getVisitor(ip string) *rate.Limiter {
rl.mu.Lock()
defer rl.mu.Unlock()
v, exists := rl.visitors[ip]
if !exists {
limiter := rate.NewLimiter(rl.rate, rl.burst)
rl.visitors[ip] = &visitor{limiter: limiter, lastSeen: time.Now()}
return limiter
}
v.lastSeen = time.Now()
return v.limiter
}
func (rl *RateLimiter) cleanupVisitors() {
for {
time.Sleep(rl.cleanup)
rl.mu.Lock()
for ip, v := range rl.visitors {
if time.Since(v.lastSeen) > rl.cleanup {
delete(rl.visitors, ip)
}
}
rl.mu.Unlock()
}
}
func (rl *RateLimiter) Limit(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
ip := r.RemoteAddr
limiter := rl.getVisitor(ip)
if !limiter.Allow() {
http.Error(w, "Rate limit exceeded", http.StatusTooManyRequests)
return
}
next.ServeHTTP(w, r)
})
}
""",
},
]
# =============================================================================
# EVALUATION METRICS
# =============================================================================
@dataclass
class AccuracyResult:
"""Ground truth accuracy measurement."""
exact_match: bool
f1_score: float
contains_answer: bool
def compute_f1(prediction: str, ground_truth: str) -> float:
"""Compute token-level F1 score."""
pred_tokens = set(prediction.lower().split())
truth_tokens = set(ground_truth.lower().split())
if not pred_tokens or not truth_tokens:
return 0.0
common = pred_tokens & truth_tokens
if not common:
return 0.0
precision = len(common) / len(pred_tokens)
recall = len(common) / len(truth_tokens)
return 2 * precision * recall / (precision + recall)
def evaluate_answer(prediction: str, ground_truth: str) -> AccuracyResult:
"""Evaluate prediction against ground truth."""
pred_lower = prediction.lower().strip()
truth_lower = ground_truth.lower().strip()
return AccuracyResult(
exact_match=pred_lower == truth_lower,
f1_score=compute_f1(prediction, ground_truth),
contains_answer=truth_lower in pred_lower,
)
# =============================================================================
# MIXED CONTENT SCENARIOS
# =============================================================================
@dataclass
class Scenario:
"""A test scenario with mixed content types."""
name: str
description: str
tool_outputs: list[dict] # Simulated tool outputs
question: str
ground_truth: str | None = None
validation_fn: Any = None # Custom validation function
def create_sre_scenario() -> Scenario:
"""
Real SRE incident scenario with mixed content:
- Kubernetes events (structured)
- Application logs (semi-structured)
- Stack traces (code)
- Metrics JSON (data)
"""
logs = load_real_logs()
return Scenario(
name="SRE Incident Investigation",
description="Debug a production outage using mixed log types",
tool_outputs=[
{
"tool": "get_kubernetes_events",
"result": logs[1]["content"], # K8s events
},
{
"tool": "get_application_logs",
"result": logs[0]["content"], # Java Spring logs
},
{
"tool": "get_error_details",
"result": logs[2]["content"], # Python traceback
},
{
"tool": "get_metrics",
"result": json.dumps(
{
"cpu_percent": [45, 47, 52, 89, 95, 98, 99, 99],
"memory_mb": [2048, 2100, 2200, 3500, 3800, 3950, 4000, 4000],
"request_latency_p99_ms": [50, 55, 60, 250, 800, 1500, 2000, 2500],
"error_rate_percent": [0.1, 0.1, 0.2, 5.0, 15.0, 25.0, 30.0, 35.0],
"timestamps": [
"14:20",
"14:25",
"14:30",
"14:35",
"14:40",
"14:45",
"14:50",
"14:55",
],
},
indent=2,
),
},
],
question="What is the root cause of this outage? What service is affected and what is the specific error?",
ground_truth="connection pool timeout / database connection exhaustion",
validation_fn=lambda r: any(
term in r.lower()
for term in [
"connection pool",
"timeout",
"database",
"pool error",
"acquire connection",
]
),
)
def create_code_review_scenario() -> Scenario:
"""
Real code review scenario with mixed content:
- Actual code (Python, TypeScript, Go)
- Code diff
- Review comments
"""
code_samples = load_real_code_samples()
return Scenario(
name="Code Review Analysis",
description="Review code across multiple languages and identify patterns",
tool_outputs=[
{
"tool": "get_file_contents",
"file": code_samples[0]["file"],
"result": code_samples[0]["content"],
},
{
"tool": "get_file_contents",
"file": code_samples[1]["file"],
"result": code_samples[1]["content"],
},
{
"tool": "get_file_contents",
"file": code_samples[2]["file"],
"result": code_samples[2]["content"],
},
{
"tool": "get_review_comments",
"result": json.dumps(
[
{
"file": "auth/middleware.py",
"line": 25,
"comment": "Should we add rate limiting here?",
},
{
"file": "hooks/useAsync.ts",
"line": 42,
"comment": "Memory leak risk if component unmounts during fetch",
},
{
"file": "middleware/ratelimit.go",
"line": 55,
"comment": "Consider using sync.Map for better concurrent performance",
},
],
indent=2,
),
},
],
question="What authentication patterns are used across these files? Are there any security concerns?",
ground_truth="JWT Bearer token authentication",
validation_fn=lambda r: any(
term in r.lower() for term in ["jwt", "bearer", "token", "authentication"]
),
)
def create_research_scenario(hotpot_samples: list[dict]) -> Scenario | None:
"""
Real research scenario using HotpotQA data.
Multi-hop reasoning with ground truth answers.
"""
if not hotpot_samples:
return None
sample = hotpot_samples[0]
return Scenario(
name="Research Question Answering",
description="Answer multi-hop question from Wikipedia passages",
tool_outputs=[
{
"tool": "search_wikipedia",
"query": sample["question"],
"result": sample["context"],
},
],
question=sample["question"],
ground_truth=sample["ground_truth"],
validation_fn=lambda r: sample["ground_truth"].lower() in r.lower(),
)
# =============================================================================
# MAIN EVALUATION HARNESS
# =============================================================================
@dataclass
class EvalResult:
"""Result from a single evaluation run."""
scenario_name: str
mode: str # "baseline" or "headroom"
tokens_before: int
tokens_after: int
compression_ratio: float
accuracy_preserved: bool
f1_score: float
latency_ms: float
response: str
def run_scenario_with_headroom(
scenario: Scenario,
model_id: str = "claude-sonnet-4-20250514",
) -> tuple[EvalResult, EvalResult]:
"""Run a scenario with and without Headroom, measure accuracy."""
from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.tools import tool
from headroom.integrations.agno import HeadroomAgnoModel
# Create tools that return our scenario data
tool_data = {t["tool"]: t["result"] for t in scenario.tool_outputs}
@tool(name="search_tool")
def search_tool(query: str) -> str:
"""Search for information."""
# Return all tool outputs concatenated (simulating multiple tool calls)
return "\n\n---\n\n".join(tool_data.values())
# Build the full context
full_context = "\n\n---\n\n".join(tool_data.values())
# Estimate tokens (rough)
baseline_tokens = len(full_context) // 4
# Run with Headroom
base_model = Claude(id=model_id)
headroom_model = HeadroomAgnoModel(wrapped_model=base_model)
agent = Agent(model=headroom_model, tools=[search_tool], markdown=True)
prompt = f"""Based on the following information from various tools:
{full_context}
Question: {scenario.question}
Provide a clear, specific answer."""
start = time.time()
response = agent.run(prompt)
response_text = response.content if hasattr(response, "content") else str(response)
latency = (time.time() - start) * 1000
# Get Headroom stats
stats = headroom_model.get_savings_summary()
tokens_after = stats.get("total_tokens_after", baseline_tokens)
tokens_before = stats.get("total_tokens_before", baseline_tokens)
# Evaluate accuracy
if scenario.ground_truth:
accuracy = evaluate_answer(response_text, scenario.ground_truth)
accuracy_preserved = accuracy.contains_answer or accuracy.f1_score > 0.5
f1 = accuracy.f1_score
elif scenario.validation_fn:
accuracy_preserved = scenario.validation_fn(response_text)
f1 = 1.0 if accuracy_preserved else 0.0
else:
accuracy_preserved = True
f1 = 1.0
compression_ratio = (tokens_before - tokens_after) / tokens_before if tokens_before > 0 else 0
baseline_result = EvalResult(
scenario_name=scenario.name,
mode="baseline",
tokens_before=tokens_before,
tokens_after=tokens_before, # No compression for baseline
compression_ratio=0.0,
accuracy_preserved=True, # Baseline is reference
f1_score=1.0,
latency_ms=0, # Not measured for baseline
response="(baseline - not run separately)",
)
headroom_result = EvalResult(
scenario_name=scenario.name,
mode="headroom",
tokens_before=tokens_before,
tokens_after=tokens_after,
compression_ratio=compression_ratio,
accuracy_preserved=accuracy_preserved,
f1_score=f1,
latency_ms=latency,
response=response_text[:500],
)
return baseline_result, headroom_result
def main():
"""Run comprehensive evaluation."""
print("\n" + "=" * 70)
print(" COMPREHENSIVE HEADROOM EVALUATION")
print(" Real Data | Real Accuracy | Mixed Content")
print("=" * 70)
# Check for API key
if not os.environ.get("ANTHROPIC_API_KEY"):
print("\n ERROR: ANTHROPIC_API_KEY environment variable required")
print(" Set it and re-run: export ANTHROPIC_API_KEY=sk-ant-...")
return
# Load real data
print("\n Loading real datasets...")
bfcl_samples = load_bfcl_samples(5)
print(f" BFCL samples: {len(bfcl_samples)}")
hotpot_samples = load_hotpotqa_samples(5)
print(f" HotpotQA samples: {len(hotpot_samples)}")
# Create scenarios
print("\n Creating test scenarios...")
scenarios = [
create_sre_scenario(),
create_code_review_scenario(),
]
research_scenario = create_research_scenario(hotpot_samples)
if research_scenario:
scenarios.append(research_scenario)
print(f" Total scenarios: {len(scenarios)}")
# Run evaluation
results = []
for scenario in scenarios:
print(f"\n Running: {scenario.name}")
print(f" {scenario.description}")
try:
baseline, headroom = run_scenario_with_headroom(scenario)
results.append((baseline, headroom))
print(
f" Tokens: {headroom.tokens_before:,}{headroom.tokens_after:,} ({headroom.compression_ratio:.1%} saved)"
)
print(f" Accuracy preserved: {'' if headroom.accuracy_preserved else ''}")
print(f" F1 score: {headroom.f1_score:.2f}")
except Exception as e:
print(f" ERROR: {e}")
# Summary
print("\n" + "=" * 70)
print(" SUMMARY")
print("=" * 70)
if results:
total_before = sum(h.tokens_before for _, h in results)
total_after = sum(h.tokens_after for _, h in results)
avg_compression = (total_before - total_after) / total_before if total_before > 0 else 0
accuracy_rate = sum(1 for _, h in results if h.accuracy_preserved) / len(results)
avg_f1 = sum(h.f1_score for _, h in results) / len(results)
print(f"""
Scenarios tested: {len(results)}
Total tokens before: {total_before:,}
Total tokens after: {total_after:,}
Average compression: {avg_compression:.1%}
Accuracy preserved: {accuracy_rate:.1%}
Average F1 score: {avg_f1:.2f}
""")
# Save results
output = {
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
"scenarios": [
{
"name": h.scenario_name,
"tokens_before": h.tokens_before,
"tokens_after": h.tokens_after,
"compression_ratio": h.compression_ratio,
"accuracy_preserved": h.accuracy_preserved,
"f1_score": h.f1_score,
}
for _, h in results
],
}
output_file = Path(__file__).parent / "comprehensive_eval_results.json"
with open(output_file, "w") as f:
json.dump(output, f, indent=2)
print(f" Results saved to: {output_file}")
print("=" * 70 + "\n")
if __name__ == "__main__":
main()

View file

@ -0,0 +1,214 @@
#!/usr/bin/env python3
"""
Groq-Specific Reasoning Test
Tests Groq model with reasoning=True, both with and without Headroom.
This isolates whether the issue is Groq-specific or Headroom-specific.
"""
import json
import os
import sys
import traceback
# Enable Agno debugging
os.environ["AGNO_DEBUG"] = "true"
from agno.agent import Agent
from agno.models.groq import Groq
from agno.tools import tool
# Check for API key
GROQ_API_KEY = os.environ.get("GROQ_API_KEY")
if not GROQ_API_KEY:
print("ERROR: GROQ_API_KEY environment variable required")
sys.exit(1)
# =============================================================================
# SIMPLE TOOLS
# =============================================================================
@tool(name="get_weather")
def get_weather(city: str) -> str:
"""Get weather for a city.
Args:
city: City name
Returns:
Weather information as JSON
"""
print(f"[TOOL] get_weather called with: {city}")
return json.dumps(
{"city": city, "temperature": "72°F", "conditions": "Sunny", "humidity": "45%"}
)
@tool(name="get_time")
def get_time(timezone: str = "UTC") -> str:
"""Get current time in a timezone.
Args:
timezone: Timezone name
Returns:
Current time
"""
print(f"[TOOL] get_time called with: {timezone}")
return json.dumps({"timezone": timezone, "time": "14:30:00", "date": "2025-01-19"})
# =============================================================================
# TEST FUNCTIONS
# =============================================================================
def test_groq(use_headroom: bool, use_reasoning: bool, model_id: str = "llama-3.3-70b-versatile"):
"""Test Groq with specific configuration."""
label = f"Groq {'+ Headroom' if use_headroom else 'Direct'}, reasoning={use_reasoning}, model={model_id}"
print(f"\n{'#' * 70}")
print(f"# TEST: {label}")
print(f"{'#' * 70}")
try:
# Create the model
if use_headroom:
from headroom.integrations.agno import HeadroomAgnoModel
base_model = Groq(id=model_id)
model = HeadroomAgnoModel(wrapped_model=base_model)
print("[SETUP] Created HeadroomAgnoModel wrapping Groq")
else:
model = Groq(id=model_id)
print("[SETUP] Created Groq model directly")
# Create the agent
agent = Agent(
model=model,
tools=[get_weather, get_time],
reasoning=use_reasoning,
markdown=True,
debug_mode=True,
)
print(f"[SETUP] Created Agent with reasoning={use_reasoning}")
# Simple question
question = "What's the weather in San Francisco and what time is it there?"
print(f"[INPUT] Question: {question}")
# Run the agent
print("[RUN] Starting agent.run()...")
response = agent.run(question)
# Extract response
if hasattr(response, "content") and response.content is not None:
response_text = response.content
elif response is not None:
response_text = str(response)
else:
response_text = "(No response content)"
print(f"[OUTPUT] Response length: {len(response_text)} chars")
print(f"[OUTPUT] Response preview: {response_text[:300]}...")
# Get Headroom stats if available
if use_headroom and hasattr(model, "get_savings_summary"):
stats = model.get_savings_summary()
print(f"[HEADROOM] Stats: {stats}")
return {
"success": True,
"label": label,
"response": response_text[:500],
}
except Exception as e:
error_msg = str(e)
tb = traceback.format_exc()
print(f"[ERROR] Exception: {error_msg}")
print(f"[ERROR] Traceback:\n{tb}")
return {
"success": False,
"label": label,
"error": error_msg,
"traceback": tb,
}
def run_all_tests():
"""Run all Groq test combinations."""
print("\n" + "=" * 70)
print("GROQ REASONING TEST")
print("=" * 70)
print(f"GROQ_API_KEY: {'SET' if GROQ_API_KEY else 'NOT SET'}")
print("=" * 70)
results = []
# Test with llama-3.3-70b-versatile (most capable)
model_id = "llama-3.3-70b-versatile"
test_cases = [
# (use_headroom, use_reasoning)
(False, False), # Baseline: Groq direct, no reasoning
(False, True), # Groq direct, with reasoning
(True, False), # Groq + Headroom, no reasoning
(True, True), # Groq + Headroom, with reasoning <-- This is what fails for user
]
for use_headroom, use_reasoning in test_cases:
result = test_groq(use_headroom, use_reasoning, model_id)
results.append(result)
print(f"\nResult: {'✅ SUCCESS' if result['success'] else '❌ FAILED'}")
if not result["success"]:
print(f"Error: {result.get('error', 'Unknown')[:200]}")
# Summary
print("\n" + "=" * 70)
print("SUMMARY")
print("=" * 70)
for result in results:
status = "✅ PASS" if result["success"] else "❌ FAIL"
print(f"{status} - {result['label']}")
if not result["success"]:
print(f" Error: {result.get('error', 'Unknown')[:100]}")
print("\n" + "=" * 70)
# Analysis
print("\nANALYSIS:")
# Check if Groq + reasoning fails without Headroom
groq_direct_reasoning = next(
(r for r in results if "Direct" in r["label"] and "reasoning=True" in r["label"]), None
)
groq_headroom_reasoning = next(
(r for r in results if "Headroom" in r["label"] and "reasoning=True" in r["label"]), None
)
if groq_direct_reasoning and not groq_direct_reasoning["success"]:
print("⚠️ Groq + reasoning=True fails WITHOUT Headroom!")
print(" This is an Agno/Groq bug, NOT a Headroom issue.")
if (
groq_direct_reasoning
and groq_direct_reasoning["success"]
and groq_headroom_reasoning
and not groq_headroom_reasoning["success"]
):
print("⚠️ Groq + reasoning=True works without Headroom but FAILS with Headroom!")
print(" This IS a Headroom issue that needs investigation.")
if all(r["success"] for r in results):
print("✅ All tests passed! No issues found.")
return results
if __name__ == "__main__":
run_all_tests()

View file

@ -0,0 +1,759 @@
#!/usr/bin/env python3
"""
Comprehensive Integration Test for IntelligentContextManager
This test runs WITHOUT MOCKS - it uses real API calls to verify:
1. IntelligentContextManager is properly wired into the pipeline
2. COMPRESS_FIRST strategy works (deeper compression before dropping)
3. SUMMARIZE strategy works (progressive summarization)
4. DROP_BY_SCORE strategy works (semantic scoring)
5. Token savings are real and significant
Requirements:
- ANTHROPIC_API_KEY environment variable
- Real API calls will be made
"""
import json
import os
import sys
import time
from dataclasses import dataclass
# Check for API key early
API_KEY = os.environ.get("ANTHROPIC_API_KEY")
if not API_KEY:
print("ERROR: ANTHROPIC_API_KEY environment variable required")
print("Usage: ANTHROPIC_API_KEY=sk-... python examples/intelligent_context_integration_test.py")
sys.exit(1)
from anthropic import Anthropic # noqa: E402
from headroom import AnthropicProvider, HeadroomClient # noqa: E402
from headroom.config import ( # noqa: E402
HeadroomConfig,
IntelligentContextConfig,
)
from headroom.transforms import IntelligentContextManager # noqa: E402
from headroom.transforms.pipeline import TransformPipeline # noqa: E402
# =============================================================================
# TEST DATA - Realistic tool outputs that benefit from intelligent compression
# =============================================================================
def generate_large_search_results(count: int = 100) -> str:
"""Generate realistic search results with varying relevance."""
results = []
for i in range(count):
# Some results are clearly important (errors, high scores)
if i == 42:
result = {
"id": i,
"title": "CRITICAL: Memory leak in worker pool causing OOM",
"score": 0.98,
"type": "error",
"content": "Worker threads are not being released after task completion. "
"Stack trace shows accumulation in ThreadPoolExecutor.",
"metadata": {"severity": "critical", "affected_users": 1247},
}
elif i == 17:
result = {
"id": i,
"title": "Performance regression in v2.3.1 release",
"score": 0.95,
"type": "bug",
"content": "Response times increased 3x after the latest deploy. "
"Profiling shows bottleneck in database connection pooling.",
"metadata": {"severity": "high", "p99_latency_ms": 2340},
}
else:
result = {
"id": i,
"title": f"Search result #{i}: {'Feature request' if i % 3 == 0 else 'Documentation update'}",
"score": 0.3 + (0.5 * (1 - i / count)), # Decreasing relevance
"type": "info",
"content": f"This is search result {i} with standard content. "
f"Contains typical information that may or may not be relevant.",
"metadata": {"views": 100 + i * 10, "last_updated": f"2024-01-{(i % 28) + 1:02d}"},
}
results.append(result)
return json.dumps(results, indent=2)
def generate_log_entries(count: int = 200) -> str:
"""Generate realistic log entries with some errors."""
entries = []
for i in range(count):
if i == 87:
entry = {
"timestamp": f"2024-01-15T10:{i % 60:02d}:00Z",
"level": "ERROR",
"service": "worker-pool",
"message": "OutOfMemoryError: Java heap space exhausted",
"stack_trace": "java.lang.OutOfMemoryError: Java heap space\n"
" at WorkerPool.execute(WorkerPool.java:234)\n"
" at TaskRunner.run(TaskRunner.java:89)",
"context": {"heap_used": "7.8GB", "heap_max": "8GB", "thread_count": 847},
}
elif i == 143:
entry = {
"timestamp": f"2024-01-15T10:{i % 60:02d}:00Z",
"level": "ERROR",
"service": "database",
"message": "Connection pool exhausted - all 100 connections in use",
"context": {"active_connections": 100, "waiting_requests": 342},
}
else:
entry = {
"timestamp": f"2024-01-15T10:{i % 60:02d}:00Z",
"level": "INFO",
"service": ["api", "auth", "worker", "cache"][i % 4],
"message": f"Request processed successfully (id={i})",
"context": {"latency_ms": 50 + (i % 100), "status": 200},
}
entries.append(entry)
return json.dumps(entries, indent=2)
def generate_code_analysis(file_count: int = 50) -> str:
"""Generate code analysis results."""
files = []
for i in range(file_count):
if i == 23:
file_result = {
"path": "src/worker.py",
"issues": [
{
"line": 234,
"type": "memory_leak",
"severity": "critical",
"message": "Thread pool executor not properly shutdown",
},
{
"line": 287,
"type": "resource_leak",
"severity": "high",
"message": "Database connection not closed in finally block",
},
],
"metrics": {"complexity": 45, "lines": 523, "test_coverage": 0.23},
}
else:
file_result = {
"path": f"src/module_{i}.py",
"issues": [],
"metrics": {"complexity": 5 + (i % 10), "lines": 100 + i * 5, "test_coverage": 0.8},
}
files.append(file_result)
return json.dumps(files, indent=2)
# =============================================================================
# TEST HELPERS
# =============================================================================
@dataclass
class TestResult:
name: str
success: bool
tokens_before: int
tokens_after: int
tokens_saved: int
savings_percent: float
strategy_used: str
duration_ms: float
error: str | None = None
SYSTEM_PROMPT = (
"You are a helpful assistant that analyzes data and provides insights. "
"When given search results or logs, identify the most important items "
"and summarize key findings."
)
def create_test_messages(tool_output: str, question: str) -> list[dict]:
"""Create a realistic conversation with tool output (no system message - passed separately)."""
return [
{"role": "user", "content": "Search for any critical issues in our system."},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_search_001",
"type": "function",
"function": {
"name": "search_issues",
"arguments": '{"query": "critical issues errors"}',
},
}
],
},
{"role": "tool", "tool_call_id": "call_search_001", "content": tool_output},
{"role": "user", "content": question},
]
def create_headroom_client(
config: HeadroomConfig, model_context_limits: dict | None = None
) -> HeadroomClient:
"""Create a HeadroomClient with the given config."""
base_client = Anthropic(api_key=API_KEY)
provider = AnthropicProvider()
return HeadroomClient(
original_client=base_client,
provider=provider,
default_mode="optimize",
config=config,
model_context_limits=model_context_limits,
)
# =============================================================================
# TESTS
# =============================================================================
def test_intelligent_context_wired_in_pipeline():
"""Test that IntelligentContextManager is properly wired into the pipeline."""
print("\n" + "=" * 70)
print("TEST: IntelligentContextManager wired into pipeline")
print("=" * 70)
# Create config with intelligent context enabled
config = HeadroomConfig()
config.intelligent_context = IntelligentContextConfig(
enabled=True,
use_importance_scoring=True,
compress_threshold=0.10,
summarize_threshold=0.25,
)
# Disable rolling window since intelligent context is enabled
config.rolling_window.enabled = False
# Create pipeline
provider = AnthropicProvider()
pipeline = TransformPipeline(config, provider=provider)
# Check that IntelligentContextManager is in the transforms
icm_found = False
for transform in pipeline.transforms:
if isinstance(transform, IntelligentContextManager):
icm_found = True
break
if icm_found:
print("✅ IntelligentContextManager found in pipeline transforms")
return TestResult(
name="Pipeline Wiring",
success=True,
tokens_before=0,
tokens_after=0,
tokens_saved=0,
savings_percent=0,
strategy_used="N/A",
duration_ms=0,
)
else:
print("❌ IntelligentContextManager NOT found in pipeline!")
print(f" Transforms in pipeline: {[t.name for t in pipeline.transforms]}")
return TestResult(
name="Pipeline Wiring",
success=False,
tokens_before=0,
tokens_after=0,
tokens_saved=0,
savings_percent=0,
strategy_used="N/A",
duration_ms=0,
error="IntelligentContextManager not in pipeline",
)
def test_compress_first_strategy():
"""Test COMPRESS_FIRST strategy - deeper compression before dropping."""
print("\n" + "=" * 70)
print("TEST: COMPRESS_FIRST strategy")
print("=" * 70)
# Generate large tool output
search_results = generate_large_search_results(100)
messages = create_test_messages(
search_results, "What are the most critical issues? Summarize the top problems."
)
print(f"Tool output size: {len(search_results):,} chars (~{len(search_results) // 4:,} tokens)")
# Create config with intelligent context
config = HeadroomConfig()
config.intelligent_context = IntelligentContextConfig(
enabled=True,
use_importance_scoring=True,
compress_threshold=0.50, # High threshold to trigger COMPRESS_FIRST
keep_last_turns=2,
)
config.rolling_window.enabled = False
config.smart_crusher.enabled = True # Enable smart crushing for COMPRESS_FIRST
start_time = time.time()
try:
client = create_headroom_client(config)
# Get optimization result via simulate (on messages API)
result = client.messages.simulate(
messages=messages,
model="claude-sonnet-4-20250514",
system=SYSTEM_PROMPT,
)
duration_ms = (time.time() - start_time) * 1000
tokens_saved = result.tokens_before - result.tokens_after
savings_pct = (tokens_saved / result.tokens_before * 100) if result.tokens_before > 0 else 0
print(f"✅ Tokens before: {result.tokens_before:,}")
print(f"✅ Tokens after: {result.tokens_after:,}")
print(f"✅ Tokens saved: {tokens_saved:,} ({savings_pct:.1f}%)")
print(f"✅ Transforms: {result.transforms}")
return TestResult(
name="COMPRESS_FIRST",
success=True,
tokens_before=result.tokens_before,
tokens_after=result.tokens_after,
tokens_saved=tokens_saved,
savings_percent=savings_pct,
strategy_used="compress_first"
if any("compress" in t.lower() for t in result.transforms)
else "smart_crusher",
duration_ms=duration_ms,
)
except Exception as e:
duration_ms = (time.time() - start_time) * 1000
print(f"❌ Error: {e}")
import traceback
traceback.print_exc()
return TestResult(
name="COMPRESS_FIRST",
success=False,
tokens_before=0,
tokens_after=0,
tokens_saved=0,
savings_percent=0,
strategy_used="error",
duration_ms=duration_ms,
error=str(e),
)
def test_drop_by_score_strategy():
"""Test DROP_BY_SCORE strategy - semantic scoring for message importance."""
print("\n" + "=" * 70)
print("TEST: DROP_BY_SCORE strategy (over budget scenario)")
print("=" * 70)
# Create a VERY long conversation that will definitely exceed limits
# System message passed separately to Anthropic API
messages = []
# Add many tool calls and responses to exceed context
for i in range(20):
messages.append({"role": "user", "content": f"Search for issue category {i}"})
messages.append(
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": f"call_{i:03d}",
"type": "function",
"function": {"name": "search", "arguments": f'{{"query": "category {i}"}}'},
}
],
}
)
# Large tool output
messages.append(
{
"role": "tool",
"tool_call_id": f"call_{i:03d}",
"content": generate_log_entries(50), # 50 log entries per call
}
)
# Final question
messages.append(
{"role": "user", "content": "Based on all the searches, what are the critical issues?"}
)
print(f"Conversation: {len(messages)} messages")
# Create config with intelligent context and SMALL context limit to force dropping
config = HeadroomConfig()
config.intelligent_context = IntelligentContextConfig(
enabled=True,
use_importance_scoring=True,
compress_threshold=0.05, # Low threshold to skip COMPRESS_FIRST
keep_last_turns=2,
output_buffer_tokens=4000,
)
config.rolling_window.enabled = False
config.smart_crusher.enabled = True
start_time = time.time()
try:
client = create_headroom_client(
config,
# Use a small context limit to force dropping
model_context_limits={"claude-sonnet-4-20250514": 20000},
)
result = client.messages.simulate(
messages=messages,
model="claude-sonnet-4-20250514",
system="You are a helpful assistant that analyzes data.",
)
duration_ms = (time.time() - start_time) * 1000
tokens_saved = result.tokens_before - result.tokens_after
savings_pct = (tokens_saved / result.tokens_before * 100) if result.tokens_before > 0 else 0
print(f"✅ Tokens before: {result.tokens_before:,}")
print(f"✅ Tokens after: {result.tokens_after:,}")
print(f"✅ Tokens saved: {tokens_saved:,} ({savings_pct:.1f}%)")
print(f"✅ Transforms: {result.transforms}")
# Check if intelligent_cap was applied (dropping happened)
dropped = any("intelligent_cap" in t for t in result.transforms)
strategy = "drop_by_score" if dropped else "compress_only"
return TestResult(
name="DROP_BY_SCORE",
success=True,
tokens_before=result.tokens_before,
tokens_after=result.tokens_after,
tokens_saved=tokens_saved,
savings_percent=savings_pct,
strategy_used=strategy,
duration_ms=duration_ms,
)
except Exception as e:
duration_ms = (time.time() - start_time) * 1000
print(f"❌ Error: {e}")
import traceback
traceback.print_exc()
return TestResult(
name="DROP_BY_SCORE",
success=False,
tokens_before=0,
tokens_after=0,
tokens_saved=0,
savings_percent=0,
strategy_used="error",
duration_ms=duration_ms,
error=str(e),
)
def test_real_api_call_with_optimization():
"""Test a real API call with intelligent context optimization."""
print("\n" + "=" * 70)
print("TEST: Real API call with optimization")
print("=" * 70)
# Create a long conversation (no tool calls - simpler for real API)
# This simulates a multi-turn conversation that benefits from compression
messages = []
# Add many conversation turns with verbose content
for i in range(15):
messages.append(
{
"role": "user",
"content": f"Tell me about topic {i}. Please provide detailed information including "
f"history, current state, key concepts, and important considerations. "
f"I want comprehensive coverage of all aspects related to topic {i}.",
}
)
messages.append(
{
"role": "assistant",
"content": f"Here's detailed information about topic {i}:\n\n"
f"**History**: Topic {i} has a rich history spanning many decades. "
f"It originated in the early period and evolved through various phases. "
f"Key milestones include development A, breakthrough B, and innovation C.\n\n"
f"**Current State**: Today, topic {i} is widely recognized as important. "
f"Modern applications include X, Y, and Z. The field continues to evolve.\n\n"
f"**Key Concepts**: Understanding topic {i} requires grasping concepts like "
f"principle 1, methodology 2, and framework 3. These form the foundation.\n\n"
f"**Considerations**: When working with topic {i}, consider factors such as "
f"constraint A, limitation B, and opportunity C. Best practices recommend "
f"approach D for optimal results.",
}
)
# Final question
messages.append(
{
"role": "user",
"content": "Based on everything we discussed, what are the 3 most important takeaways?",
}
)
print(f"Conversation: {len(messages)} messages")
config = HeadroomConfig()
config.intelligent_context = IntelligentContextConfig(
enabled=True,
use_importance_scoring=True,
)
config.rolling_window.enabled = False
config.smart_crusher.enabled = True
start_time = time.time()
try:
client = create_headroom_client(config)
# Make actual API call using the Anthropic-style API
response = client.messages.create(
model="claude-sonnet-4-20250514",
messages=messages,
system="You are a helpful assistant. Be concise in your responses.",
max_tokens=300,
)
duration_ms = (time.time() - start_time) * 1000
# Extract response content
if hasattr(response, "content") and response.content:
if isinstance(response.content, list):
response_text = response.content[0].text if response.content else ""
else:
response_text = str(response.content)
else:
response_text = str(response)
print("✅ API call successful")
print(f"✅ Response length: {len(response_text)} chars")
print(f"✅ Response preview: {response_text[:200]}...")
print(f"✅ Duration: {duration_ms:.0f}ms")
# Get session stats from internal tracking
stats = client._session_stats
tokens_before = stats.get("tokens", {}).get("input_before", 0)
tokens_after = stats.get("tokens", {}).get("input_after", 0)
tokens_saved = tokens_before - tokens_after
savings_pct = (tokens_saved / tokens_before * 100) if tokens_before > 0 else 0
print(f"✅ Session tokens before: {tokens_before:,}")
print(f"✅ Session tokens after: {tokens_after:,}")
print(f"✅ Session tokens saved: {tokens_saved:,} ({savings_pct:.1f}%)")
return TestResult(
name="Real API Call",
success=True,
tokens_before=tokens_before,
tokens_after=tokens_after,
tokens_saved=tokens_saved,
savings_percent=savings_pct,
strategy_used="intelligent_context",
duration_ms=duration_ms,
)
except Exception as e:
duration_ms = (time.time() - start_time) * 1000
print(f"❌ Error: {e}")
import traceback
traceback.print_exc()
return TestResult(
name="Real API Call",
success=False,
tokens_before=0,
tokens_after=0,
tokens_saved=0,
savings_percent=0,
strategy_used="error",
duration_ms=duration_ms,
error=str(e),
)
def test_comparison_rolling_window_vs_intelligent():
"""Compare RollingWindow vs IntelligentContextManager on same data."""
print("\n" + "=" * 70)
print("TEST: RollingWindow vs IntelligentContextManager comparison")
print("=" * 70)
# Create test data
search_results = generate_large_search_results(80)
messages = create_test_messages(search_results, "Summarize the critical issues.")
results = {}
# Test with RollingWindow
print("\n--- RollingWindow ---")
config_rw = HeadroomConfig()
config_rw.rolling_window.enabled = True
config_rw.intelligent_context.enabled = False
config_rw.smart_crusher.enabled = True
try:
client_rw = create_headroom_client(config_rw)
result_rw = client_rw.messages.simulate(
messages=messages,
model="claude-sonnet-4-20250514",
system=SYSTEM_PROMPT,
)
results["rolling_window"] = {
"tokens_before": result_rw.tokens_before,
"tokens_after": result_rw.tokens_after,
"transforms": result_rw.transforms,
}
print(f"Tokens: {result_rw.tokens_before:,} -> {result_rw.tokens_after:,}")
print(f"Transforms: {result_rw.transforms}")
except Exception as e:
print(f"Error: {e}")
results["rolling_window"] = {"error": str(e)}
# Test with IntelligentContextManager
print("\n--- IntelligentContextManager ---")
config_icm = HeadroomConfig()
config_icm.rolling_window.enabled = False
config_icm.intelligent_context = IntelligentContextConfig(enabled=True)
config_icm.smart_crusher.enabled = True
try:
client_icm = create_headroom_client(config_icm)
result_icm = client_icm.messages.simulate(
messages=messages,
model="claude-sonnet-4-20250514",
system=SYSTEM_PROMPT,
)
results["intelligent_context"] = {
"tokens_before": result_icm.tokens_before,
"tokens_after": result_icm.tokens_after,
"transforms": result_icm.transforms,
}
print(f"Tokens: {result_icm.tokens_before:,} -> {result_icm.tokens_after:,}")
print(f"Transforms: {result_icm.transforms}")
except Exception as e:
print(f"Error: {e}")
results["intelligent_context"] = {"error": str(e)}
# Compare
print("\n--- Comparison ---")
if "error" not in results.get("rolling_window", {}) and "error" not in results.get(
"intelligent_context", {}
):
rw_saved = (
results["rolling_window"]["tokens_before"] - results["rolling_window"]["tokens_after"]
)
icm_saved = (
results["intelligent_context"]["tokens_before"]
- results["intelligent_context"]["tokens_after"]
)
print(f"RollingWindow saved: {rw_saved:,} tokens")
print(f"IntelligentContext saved: {icm_saved:,} tokens")
if icm_saved >= rw_saved:
print(f"✅ IntelligentContextManager saved {icm_saved - rw_saved:,} MORE tokens!")
else:
print(f"⚠️ RollingWindow saved {rw_saved - icm_saved:,} more tokens")
return TestResult(
name="Comparison",
success=True,
tokens_before=results["intelligent_context"]["tokens_before"],
tokens_after=results["intelligent_context"]["tokens_after"],
tokens_saved=icm_saved,
savings_percent=(icm_saved / results["intelligent_context"]["tokens_before"] * 100),
strategy_used=f"ICM:{icm_saved} vs RW:{rw_saved}",
duration_ms=0,
)
else:
return TestResult(
name="Comparison",
success=False,
tokens_before=0,
tokens_after=0,
tokens_saved=0,
savings_percent=0,
strategy_used="error",
duration_ms=0,
error="One or both tests failed",
)
# =============================================================================
# MAIN
# =============================================================================
def main():
print("\n" + "=" * 70)
print("INTELLIGENT CONTEXT MANAGER - COMPREHENSIVE INTEGRATION TEST")
print("=" * 70)
print(f"API Key: {'SET' if API_KEY else 'NOT SET'}")
print("Using real API calls - NO MOCKS")
print("=" * 70)
all_results: list[TestResult] = []
# Run all tests
all_results.append(test_intelligent_context_wired_in_pipeline())
all_results.append(test_compress_first_strategy())
all_results.append(test_drop_by_score_strategy())
all_results.append(test_real_api_call_with_optimization())
all_results.append(test_comparison_rolling_window_vs_intelligent())
# Summary
print("\n" + "=" * 70)
print("SUMMARY")
print("=" * 70)
passed = 0
failed = 0
total_tokens_saved = 0
for result in all_results:
status = "✅ PASS" if result.success else "❌ FAIL"
print(f"\n{status} - {result.name}")
if result.success:
passed += 1
if result.tokens_saved > 0:
print(
f" Tokens: {result.tokens_before:,} -> {result.tokens_after:,} "
f"(saved {result.tokens_saved:,}, {result.savings_percent:.1f}%)"
)
total_tokens_saved += result.tokens_saved
print(f" Strategy: {result.strategy_used}")
if result.duration_ms > 0:
print(f" Duration: {result.duration_ms:.0f}ms")
else:
failed += 1
if result.error:
print(f" Error: {result.error[:100]}")
print("\n" + "=" * 70)
print(f"RESULTS: {passed} passed, {failed} failed")
print(f"Total tokens saved across tests: {total_tokens_saved:,}")
print("=" * 70)
return 0 if failed == 0 else 1
if __name__ == "__main__":
sys.exit(main())

View file

@ -1,336 +1,294 @@
#!/usr/bin/env python3
"""
Multi-Tool Agent Test: Diverse Data Types with Claude API
Multi-Tool Agent Demo: Headroom in Action
This test creates an agent with multiple tools returning different data types:
- GitHub: Issues, PRs, repo metadata
- ArXiv: Paper abstracts and citations
- Code Search: Source code snippets
- Database: JSON records
Shows an AI agent investigating a memory leak using 4 tools.
Headroom compresses the tool outputs while preserving critical information.
We run it WITHOUT Headroom and WITH Headroom to compare token usage.
Uses Claude API for real function calling.
Usage:
export ANTHROPIC_API_KEY=sk-ant-...
python examples/multi_tool_agent_test.py
"""
import json
import os
import time
from dataclasses import dataclass
from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.tools import tool
# Check for API key
if not os.environ.get("ANTHROPIC_API_KEY"):
raise ValueError("ANTHROPIC_API_KEY environment variable required")
# =============================================================================
# MOCK TOOL DATA - Realistic responses from various sources
# MOCK TOOL DATA
# Each tool returns many items, but only ONE is critical (the "needle")
# =============================================================================
GITHUB_ISSUES = [
{
"number": i,
"title": f"Issue #{i}: {'Memory leak in worker pool' if i == 42 else 'Feature request: ' + ['dark mode', 'API pagination', 'webhook support', 'rate limiting'][i % 4]}",
"state": "open" if i % 3 != 0 else "closed",
"author": f"user{i % 20}",
"labels": ["bug", "priority:high"] if i == 42 else ["enhancement"],
"created_at": f"2024-12-{(i % 28) + 1:02d}T10:00:00Z",
"updated_at": f"2024-12-{(i % 28) + 1:02d}T15:00:00Z",
"comments": i % 10,
"body": "Worker threads are not being released after task completion, causing memory to grow unboundedly. Stack trace attached."
if i == 42
else f"Please add support for {['dark mode', 'API pagination', 'webhook support', 'rate limiting'][i % 4]}. This would greatly improve the user experience.",
"assignees": ["maintainer1"] if i == 42 else [],
"milestone": "v2.0" if i < 20 else None,
"reactions": {"thumbs_up": 47 if i == 42 else i % 5, "thumbs_down": 0},
}
for i in range(50)
]
ARXIV_PAPERS = [
{
"id": f"2401.{i:05d}",
"title": f"{'Attention Is All You Need: Revisited' if i == 15 else ['Deep Learning for Code Generation', 'Efficient Transformers', 'Neural Architecture Search', 'Language Model Scaling'][i % 4]}",
"authors": [f"Author{j}" for j in range(3 + i % 3)],
"abstract": "We revisit the transformer architecture and propose key optimizations that reduce memory usage by 40% while maintaining accuracy. Our method introduces sparse attention patterns..."
if i == 15
else f"This paper presents a novel approach to {['code generation', 'transformer efficiency', 'neural architecture', 'model scaling'][i % 4]}. We demonstrate state-of-the-art results on benchmark datasets.",
"categories": ["cs.LG", "cs.CL"] if i == 15 else ["cs.LG"],
"published": f"2024-01-{(i % 28) + 1:02d}",
"citations": 1247 if i == 15 else i * 3,
"pdf_url": f"https://arxiv.org/pdf/2401.{i:05d}.pdf",
"comment": "Accepted at NeurIPS 2024" if i == 15 else None,
}
for i in range(30)
]
def generate_github_issues(count: int = 50, needle_at: int = 42) -> list[dict]:
"""Generate GitHub issues with a memory leak bug at needle_at."""
issues = []
for i in range(count):
if i == needle_at:
issues.append(
{
"number": i,
"title": "Memory leak in worker pool",
"state": "open",
"labels": ["bug", "priority:high"],
"body": "Worker threads not released after task completion. Memory grows unboundedly.",
"reactions": {"thumbs_up": 47},
}
)
else:
issues.append(
{
"number": i,
"title": f"Feature: {['dark mode', 'pagination', 'webhooks', 'rate limiting'][i % 4]}",
"state": "closed" if i % 3 == 0 else "open",
"labels": ["enhancement"],
"body": "Please add this feature.",
"reactions": {"thumbs_up": i % 5},
}
)
return issues
CODE_SEARCH_RESULTS = [
{
"file": f"src/{'worker.py' if i == 23 else ['utils.py', 'api.py', 'models.py', 'handlers.py'][i % 4]}",
"line": 100 + i * 10,
"content": '''def cleanup_worker(self):
"""Release worker resources - MEMORY LEAK FIX"""
def generate_code_results(count: int = 40, needle_at: int = 23) -> list[dict]:
"""Generate code search results with memory fix at needle_at."""
results = []
for i in range(count):
if i == needle_at:
results.append(
{
"file": "src/worker.py",
"line": 330,
"content": """def cleanup_worker(self):
\"\"\"Release worker resources - MEMORY LEAK FIX\"\"\"
self.thread_pool.shutdown(wait=True)
self.connections.clear()
gc.collect() # Force garbage collection'''
if i == 23
else f'''def process_{["data", "request", "model", "event"][i % 4]}(self, input):
"""Process incoming {["data", "request", "model", "event"][i % 4]}"""
result = self.transform(input)
return self.validate(result)''',
"language": "python",
"repository": "main-app",
"relevance_score": 0.98 if i == 23 else 0.7 - (i * 0.01),
"context_before": [" # Worker management", " "],
"context_after": ["", " def start_worker(self):"],
}
for i in range(40)
]
gc.collect()""",
"relevance_score": 0.98,
}
)
else:
results.append(
{
"file": f"src/{['utils', 'api', 'models', 'handlers'][i % 4]}.py",
"line": 100 + i * 10,
"content": f"def process_{['data', 'request', 'model', 'event'][i % 4]}(self): pass",
"relevance_score": 0.5 - (i * 0.01),
}
)
return results
DATABASE_RECORDS = [
{
"id": f"rec_{i:06d}",
"type": "error" if i == 17 else "info",
"timestamp": f"2024-12-15T{(i % 24):02d}:{(i % 60):02d}:00Z",
"service": "worker-pool" if i == 17 else ["api", "auth", "db", "cache"][i % 4],
"message": "OutOfMemoryError: heap space exhausted in WorkerPool.execute()"
if i == 17
else f"Operation completed: {['request processed', 'user authenticated', 'query executed', 'cache updated'][i % 4]}",
"metadata": {
"heap_used": "7.8GB" if i == 17 else f"{1 + i % 3}GB",
"heap_max": "8GB",
"thread_count": 847 if i == 17 else 50 + i % 50,
},
"stack_trace": "java.lang.OutOfMemoryError: Java heap space\n\tat WorkerPool.execute(WorkerPool.java:234)\n\tat TaskRunner.run(TaskRunner.java:89)"
if i == 17
else None,
}
for i in range(60)
]
def generate_db_logs(count: int = 60, needle_at: int = 17) -> list[dict]:
"""Generate database logs with OutOfMemoryError at needle_at."""
logs = []
for i in range(count):
if i == needle_at:
logs.append(
{
"type": "error",
"service": "worker-pool",
"message": "OutOfMemoryError: heap space exhausted",
"metadata": {"heap_used": "7.8GB", "heap_max": "8GB", "thread_count": 847},
"stack_trace": "java.lang.OutOfMemoryError at WorkerPool.execute()",
}
)
else:
logs.append(
{
"type": "info",
"service": ["api", "auth", "db", "cache"][i % 4],
"message": "Operation completed successfully",
"metadata": {"heap_used": f"{1 + i % 3}GB", "thread_count": 50 + i % 50},
}
)
return logs
def generate_arxiv_papers(count: int = 30, needle_at: int = 15) -> list[dict]:
"""Generate ArXiv papers with relevant research at needle_at."""
papers = []
for i in range(count):
if i == needle_at:
papers.append(
{
"id": f"2401.{i:05d}",
"title": "Memory Management in Worker Pool Architectures",
"abstract": "We analyze memory leak patterns in thread pools and propose cleanup strategies.",
"citations": 89,
}
)
else:
papers.append(
{
"id": f"2401.{i:05d}",
"title": f"{['Deep Learning', 'Transformers', 'Neural Arch', 'Scaling'][i % 4]} Study",
"abstract": "Novel approach with state-of-the-art results.",
"citations": i * 3,
}
)
return papers
# Generate the mock data
GITHUB_ISSUES = generate_github_issues()
CODE_RESULTS = generate_code_results()
DB_LOGS = generate_db_logs()
ARXIV_PAPERS = generate_arxiv_papers()
# =============================================================================
# TOOL DEFINITIONS
# TOOLS
# =============================================================================
@tool(name="search_github_issues")
def search_github_issues(query: str, repo: str = "main-app") -> str:
"""Search GitHub issues in a repository.
Args:
query: Search query for issues
repo: Repository name
Returns:
JSON array of matching issues
"""
def search_github_issues(query: str) -> str:
"""Search GitHub issues."""
return json.dumps(GITHUB_ISSUES, indent=2)
@tool(name="search_arxiv_papers")
def search_arxiv_papers(query: str, max_results: int = 30) -> str:
"""Search ArXiv for academic papers.
Args:
query: Search query for papers
max_results: Maximum number of results
Returns:
JSON array of matching papers
"""
return json.dumps(ARXIV_PAPERS, indent=2)
@tool(name="search_code")
def search_code(query: str, language: str = "python") -> str:
"""Search codebase for matching code.
Args:
query: Code search query
language: Programming language filter
Returns:
JSON array of code search results
"""
return json.dumps(CODE_SEARCH_RESULTS, indent=2)
def search_code(query: str) -> str:
"""Search codebase."""
return json.dumps(CODE_RESULTS, indent=2)
@tool(name="query_database")
def query_database(query: str, table: str = "logs") -> str:
"""Query the database for records.
def query_database(query: str) -> str:
"""Query database logs."""
return json.dumps(DB_LOGS, indent=2)
Args:
query: SQL-like query
table: Table to query
Returns:
JSON array of database records
"""
return json.dumps(DATABASE_RECORDS, indent=2)
@tool(name="search_arxiv")
def search_arxiv(query: str) -> str:
"""Search ArXiv papers."""
return json.dumps(ARXIV_PAPERS, indent=2)
# =============================================================================
# TEST RUNNER
# VERIFICATION
# =============================================================================
@dataclass
class TestResult:
label: str
input_tokens: int
output_tokens: int
response: str
duration_ms: float
tool_calls: int
NEEDLES = {
"GitHub Issue #42": lambda r: "memory leak" in r.lower()
and ("42" in r or "worker" in r.lower()),
"cleanup_worker() fix": lambda r: "cleanup" in r.lower() or "worker.py" in r.lower(),
"OutOfMemoryError": lambda r: "outofmemory" in r.lower() or "847" in r or "7.8" in r.lower(),
"ArXiv paper": lambda r: "paper" in r.lower()
or "research" in r.lower()
or "arxiv" in r.lower(),
}
def count_tokens_approx(text: str) -> int:
"""Approximate token count (Ollama doesn't always report tokens)."""
return len(text) // 4
def verify_response(response: str) -> dict[str, bool]:
"""Check if response found all needles."""
return {name: check(response) for name, check in NEEDLES.items()}
def run_agent_test(use_headroom: bool) -> TestResult:
"""Run the multi-tool agent test."""
label = "WITH Headroom" if use_headroom else "WITHOUT Headroom (Baseline)"
if use_headroom:
from headroom.integrations.agno import HeadroomAgnoModel
base_model = Claude(id="claude-sonnet-4-20250514")
model = HeadroomAgnoModel(wrapped_model=base_model)
else:
model = Claude(id="claude-sonnet-4-20250514")
agent = Agent(
model=model,
tools=[search_github_issues, search_arxiv_papers, search_code, query_database],
markdown=True,
)
# The question that requires searching multiple sources
question = """I'm investigating a memory leak in our application. Please:
1. Search GitHub issues for memory-related bugs
2. Search our codebase for memory leak fixes
3. Check the database logs for OutOfMemory errors
4. Find any relevant research papers about memory management in worker pools
Summarize what you find and recommend a fix."""
print(f"\n{'=' * 70}")
print(f"Running: {label}")
print(f"{'=' * 70}")
print(f"Question: {question[:100]}...")
start_time = time.time()
try:
response = agent.run(question)
response_text = response.content if hasattr(response, "content") else str(response)
except Exception as e:
response_text = f"Error: {e}"
duration_ms = (time.time() - start_time) * 1000
# Get token counts
if use_headroom and hasattr(model, "total_tokens_saved"):
summary = model.get_savings_summary()
input_tokens = summary.get("total_tokens_after", 0) # Actual tokens sent to API
tokens_before = summary.get("total_tokens_before", 0)
tokens_saved = model.total_tokens_saved
savings_pct = (tokens_saved / tokens_before * 100) if tokens_before > 0 else 0
print("\n📊 Headroom Optimization Stats:")
print(f" API requests made: {summary.get('total_requests', 0)}")
print(f" Tokens BEFORE optimization: {tokens_before:,}")
print(f" Tokens AFTER optimization: {input_tokens:,}")
print(f" Tokens SAVED: {tokens_saved:,} ({savings_pct:.1f}%)")
else:
# Estimate from data size
total_data = (
json.dumps(GITHUB_ISSUES)
+ json.dumps(ARXIV_PAPERS)
+ json.dumps(CODE_SEARCH_RESULTS)
+ json.dumps(DATABASE_RECORDS)
)
input_tokens = count_tokens_approx(total_data + question)
print(f"\nResponse preview: {response_text[:500]}...")
print(f"Duration: {duration_ms:.0f}ms")
return TestResult(
label=label,
input_tokens=input_tokens,
output_tokens=count_tokens_approx(response_text),
response=response_text,
duration_ms=duration_ms,
tool_calls=4, # We expect 4 tool calls
)
# =============================================================================
# MAIN
# =============================================================================
def main():
from headroom.integrations.agno import HeadroomAgnoModel
from headroom.pricing import estimate_cost
model_id = "claude-sonnet-4-20250514"
print("\n" + "=" * 70)
print("MULTI-TOOL AGENT TEST")
print("Testing diverse data types: GitHub, ArXiv, Code, Database")
print("Model: Claude Sonnet (claude-sonnet-4-20250514)")
print(" MULTI-TOOL AGENT DEMO")
print("=" * 70)
# Show data sizes
print("\nTool output sizes:")
print(
f" GitHub Issues: {len(json.dumps(GITHUB_ISSUES)):,} chars ({len(GITHUB_ISSUES)} items)"
)
print(f" ArXiv Papers: {len(json.dumps(ARXIV_PAPERS)):,} chars ({len(ARXIV_PAPERS)} items)")
print(
f" Code Search: {len(json.dumps(CODE_SEARCH_RESULTS)):,} chars ({len(CODE_SEARCH_RESULTS)} items)"
)
print(
f" Database Logs: {len(json.dumps(DATABASE_RECORDS)):,} chars ({len(DATABASE_RECORDS)} items)"
)
total_chars = sum(
len(json.dumps(d))
for d in [GITHUB_ISSUES, ARXIV_PAPERS, CODE_SEARCH_RESULTS, DATABASE_RECORDS]
len(json.dumps(d)) for d in [GITHUB_ISSUES, CODE_RESULTS, DB_LOGS, ARXIV_PAPERS]
)
print(f" TOTAL: {total_chars:,} chars (~{total_chars // 4:,} tokens)")
print(f"\n Tool outputs: {total_chars:,} chars across 4 tools")
print(" Needles hidden at positions: #42, #23, #17, #15")
# Run baseline (no Headroom)
print("\n" + "-" * 70)
baseline = run_agent_test(use_headroom=False)
# Create agent with Headroom
base_model = Claude(id=model_id)
model = HeadroomAgnoModel(wrapped_model=base_model)
agent = Agent(
model=model,
tools=[search_github_issues, search_code, query_database, search_arxiv],
markdown=True,
)
# Run with Headroom
print("\n" + "-" * 70)
optimized = run_agent_test(use_headroom=True)
question = """Investigate a memory leak in our application:
1. Search GitHub for memory-related issues
2. Search code for memory leak fixes
3. Check database logs for OutOfMemory errors
4. Find relevant research papers
# Final comparison
Summarize findings and recommend a fix."""
print("\n Running agent...")
start = time.time()
response = agent.run(question)
response_text = response.content if hasattr(response, "content") else str(response)
duration = time.time() - start
# Get stats from Headroom
stats = model.get_savings_summary()
tokens_before = stats["total_tokens_before"]
tokens_after = stats["total_tokens_after"]
tokens_saved = stats["total_tokens_saved"]
pct_saved = (tokens_saved / tokens_before * 100) if tokens_before > 0 else 0
# Calculate costs
cost_before = estimate_cost(model_id, input_tokens=tokens_before)
cost_after = estimate_cost(model_id, input_tokens=tokens_after)
# Verify findings
verification = verify_response(response_text)
found = sum(verification.values())
# Results
print("\n" + "=" * 70)
print("FINAL COMPARISON")
print(" RESULTS")
print("=" * 70)
print(f"""
Baseline Headroom
Tokens Sent to API: {baseline.input_tokens:>6,} {optimized.input_tokens:>6,}
Duration: {baseline.duration_ms:>6,.0f}ms {optimized.duration_ms:>6,.0f}ms
Tool Calls: {baseline.tool_calls:>6} {optimized.tool_calls:>6}
Without With
Headroom Headroom Savings
Input Tokens {tokens_before:>8,} {tokens_after:>8,} {tokens_saved:,} ({pct_saved:.0f}%)""")
if cost_before and cost_after:
cost_saved = cost_before - cost_after
print(
f" Input Cost ${cost_before:.4f} ${cost_after:.4f} ${cost_saved:.4f}"
)
print(f"""
Duration {duration:.1f}s
API Requests {stats["total_requests"]}
Needles Found {found}/4
""")
if baseline.input_tokens > optimized.input_tokens:
saved = baseline.input_tokens - optimized.input_tokens
percent = (saved / baseline.input_tokens) * 100
print(f" ✨ Tokens Saved: {saved:,} ({percent:.1f}% reduction)")
print(f" 💰 Estimated Cost Savings: {percent:.0f}% on input tokens")
for name, found_it in verification.items():
print(f" {'' if found_it else ''} {name}")
print("\n" + "=" * 70)
print("BASELINE RESPONSE (excerpt):")
print(" RESPONSE (excerpt)")
print("=" * 70)
print(baseline.response[:1500] if len(baseline.response) > 1500 else baseline.response)
print(response_text[:1500] + "..." if len(response_text) > 1500 else response_text)
print("\n" + "=" * 70)
print("HEADROOM RESPONSE (excerpt):")
print("=" * 70)
print(optimized.response[:1500] if len(optimized.response) > 1500 else optimized.response)
print(f" {pct_saved:.0f}% token reduction, {found}/4 needles found")
print("=" * 70 + "\n")
if __name__ == "__main__":

View file

@ -0,0 +1,791 @@
#!/usr/bin/env python3
"""
Quality Retention Evaluation for Intelligent Context Management
This eval verifies that when we intelligently drop/compress content,
we RETAIN critical information that the model needs to answer correctly.
Methodology:
1. NEEDLE-IN-HAYSTACK: Embed specific facts in large tool outputs
2. COMPRESS: Apply IntelligentContextManager
3. VERIFY: Ask questions requiring those facts
4. SCORE: Compare answers before/after compression
Key Metrics:
- Retention Rate: % of critical facts preserved
- Answer Accuracy: % of verification questions answered correctly
- Quality-Adjusted Savings: compression_ratio * retention_rate
Usage:
ANTHROPIC_API_KEY=sk-... python examples/quality_retention_eval.py
"""
import json
import os
import sys
from dataclasses import dataclass, field
from typing import Any
# Check for API key early
API_KEY = os.environ.get("ANTHROPIC_API_KEY")
if not API_KEY:
print("ERROR: ANTHROPIC_API_KEY environment variable required")
sys.exit(1)
from anthropic import Anthropic # noqa: E402
from headroom import AnthropicProvider, HeadroomClient # noqa: E402
from headroom.config import HeadroomConfig, IntelligentContextConfig # noqa: E402
from headroom.tokenizer import Tokenizer # noqa: E402
from headroom.tokenizers import TiktokenCounter # noqa: E402
# =============================================================================
# TEST CASE DEFINITIONS
# =============================================================================
@dataclass
class CriticalFact:
"""A fact that MUST be retained after compression."""
description: str
value: str
verification_question: str
expected_answer_contains: list[str] # Answer should contain these strings
@dataclass
class EvalTestCase:
"""A test case for quality retention evaluation."""
name: str
description: str
messages: list[dict[str, Any]]
critical_facts: list[CriticalFact]
system_prompt: str = "You are a helpful assistant. Answer questions accurately based on the information provided."
@dataclass
class EvalResult:
"""Result of evaluating a single test case."""
test_name: str
tokens_before: int
tokens_after: int
compression_ratio: float
facts_tested: int
facts_retained: int
retention_rate: float
quality_adjusted_savings: float
details: list[dict[str, Any]] = field(default_factory=list)
error: str | None = None
# =============================================================================
# TEST CASE GENERATORS
# =============================================================================
def create_search_results_with_needles() -> EvalTestCase:
"""
Test case: 100 search results with 3 critical "needles" hidden inside.
The critical facts are specific error reports that should be retained
even when most results are compressed away.
"""
results = []
# Needle 1: Critical memory error at position 42
# Needle 2: Database failure at position 17
# Needle 3: Security incident at position 73
for i in range(100):
if i == 42:
result = {
"id": f"INC-{i:04d}",
"title": "CRITICAL: OutOfMemory in production worker pool",
"severity": "P0",
"timestamp": "2024-01-15T10:42:37Z",
"server": "prod-worker-7",
"details": {
"error_type": "java.lang.OutOfMemoryError",
"heap_used_gb": 7.8,
"heap_max_gb": 8.0,
"thread_count": 847,
"affected_users": 12847,
},
"root_cause": "Thread pool executor not releasing completed tasks",
"resolution": "Restart required, fix deployed in v2.4.1",
}
elif i == 17:
result = {
"id": f"INC-{i:04d}",
"title": "Database connection pool exhausted",
"severity": "P1",
"timestamp": "2024-01-15T09:17:22Z",
"server": "db-primary-3",
"details": {
"error_type": "ConnectionPoolExhausted",
"active_connections": 500,
"max_connections": 500,
"waiting_queries": 2341,
"longest_wait_ms": 45000,
},
"root_cause": "Slow query from analytics job holding connections",
"resolution": "Killed analytics query, added connection timeout",
}
elif i == 73:
result = {
"id": f"INC-{i:04d}",
"title": "SECURITY: Unauthorized API access attempt blocked",
"severity": "P0",
"timestamp": "2024-01-15T14:23:55Z",
"source_ip": "203.0.113.42",
"details": {
"attack_type": "credential_stuffing",
"attempts": 15847,
"accounts_targeted": 892,
"accounts_compromised": 0,
"blocked_by": "rate_limiter_v2",
},
"root_cause": "Stolen credentials from third-party breach",
"resolution": "IP blocked, affected users notified to reset passwords",
}
else:
result = {
"id": f"INC-{i:04d}",
"title": f"Routine alert #{i}: {'CPU spike' if i % 3 == 0 else 'Latency increase' if i % 3 == 1 else 'Disk usage warning'}",
"severity": "P3",
"timestamp": f"2024-01-15T{10 + (i % 8):02d}:{i % 60:02d}:00Z",
"server": f"app-server-{i % 20}",
"details": {
"metric": "cpu_percent"
if i % 3 == 0
else "latency_p99"
if i % 3 == 1
else "disk_usage",
"value": 75 + (i % 20),
"threshold": 80,
"duration_minutes": 5 + (i % 10),
},
"root_cause": "Normal traffic variation",
"resolution": "Auto-resolved",
}
results.append(result)
messages = [
{"role": "user", "content": "Search for all incidents from today"},
{
"role": "assistant",
"content": [
{
"type": "tool_use",
"id": "call_search",
"name": "search_incidents",
"input": {"date": "2024-01-15"},
}
],
},
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "call_search",
"content": json.dumps(results, indent=2),
}
],
},
{
"role": "user",
"content": "What were the critical issues today? I need specific details.",
},
]
critical_facts = [
CriticalFact(
description="Memory error server identification",
value="prod-worker-7",
verification_question="Which server experienced the OutOfMemory error?",
expected_answer_contains=["prod-worker-7"],
),
CriticalFact(
description="Memory error affected users count",
value="12847",
verification_question="How many users were affected by the memory error?",
expected_answer_contains=["12847", "12,847"],
),
CriticalFact(
description="Database error connection count",
value="500 active connections",
verification_question="How many active database connections were there when the pool was exhausted?",
expected_answer_contains=["500"],
),
CriticalFact(
description="Security attack source IP",
value="203.0.113.42",
verification_question="What IP address was the source of the credential stuffing attack?",
expected_answer_contains=["203.0.113.42"],
),
CriticalFact(
description="Security attack attempt count",
value="15847 attempts",
verification_question="How many credential stuffing attempts were made?",
expected_answer_contains=["15847", "15,847"],
),
]
return EvalTestCase(
name="search_results_needles",
description="100 search results with 3 critical incidents (needles) at positions 17, 42, 73",
messages=messages,
critical_facts=critical_facts,
)
def create_log_analysis_with_needles() -> EvalTestCase:
"""
Test case: 200 log entries with critical error buried in the middle.
"""
logs = []
for i in range(200):
if i == 127:
# The critical needle - a specific error with unique identifiers
log = {
"timestamp": "2024-01-15T11:27:33.847Z",
"level": "ERROR",
"service": "payment-gateway",
"trace_id": "abc123def456",
"message": "Payment processing failed: Card declined",
"details": {
"transaction_id": "TXN-98765432",
"amount_cents": 15999,
"currency": "USD",
"error_code": "CARD_DECLINED_INSUFFICIENT_FUNDS",
"customer_id": "CUST-789012",
"retry_count": 3,
"final_status": "FAILED",
},
}
elif i == 45:
# Another needle - rate limit hit
log = {
"timestamp": "2024-01-15T10:45:12.123Z",
"level": "WARN",
"service": "api-gateway",
"message": "Rate limit exceeded for client",
"details": {
"client_id": "CLIENT-ACME-001",
"endpoint": "/api/v2/bulk-upload",
"requests_per_minute": 1500,
"limit": 1000,
"blocked_duration_seconds": 300,
},
}
else:
log = {
"timestamp": f"2024-01-15T{10 + (i % 4):02d}:{i % 60:02d}:{i % 60:02d}.{i % 1000:03d}Z",
"level": "INFO",
"service": ["api", "auth", "worker", "cache", "db"][i % 5],
"message": f"Request processed successfully (id={i})",
"details": {"latency_ms": 50 + (i % 100), "status": 200},
}
logs.append(log)
messages = [
{"role": "user", "content": "Get the logs from the last hour"},
{
"role": "assistant",
"content": [
{
"type": "tool_use",
"id": "call_logs",
"name": "get_logs",
"input": {"timerange": "1h"},
}
],
},
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "call_logs",
"content": json.dumps(logs, indent=2),
}
],
},
{"role": "user", "content": "Are there any errors or warnings I should know about?"},
]
critical_facts = [
CriticalFact(
description="Failed transaction ID",
value="TXN-98765432",
verification_question="What was the transaction ID of the failed payment?",
expected_answer_contains=["TXN-98765432"],
),
CriticalFact(
description="Payment failure amount",
value="$159.99 (15999 cents)",
verification_question="What was the amount of the failed payment in dollars?",
expected_answer_contains=["159.99", "159"],
),
CriticalFact(
description="Rate limited client",
value="CLIENT-ACME-001",
verification_question="Which client was rate limited?",
expected_answer_contains=["CLIENT-ACME-001", "ACME"],
),
CriticalFact(
description="Rate limit endpoint",
value="/api/v2/bulk-upload",
verification_question="Which API endpoint triggered the rate limit?",
expected_answer_contains=["bulk-upload", "/api/v2/bulk-upload"],
),
]
return EvalTestCase(
name="log_analysis_needles",
description="200 log entries with critical error at position 127 and warning at position 45",
messages=messages,
critical_facts=critical_facts,
)
def create_code_review_with_needles() -> EvalTestCase:
"""
Test case: Code review results with security vulnerabilities hidden in large output.
"""
files = []
for i in range(80):
if i == 23:
# Critical security vulnerability
file_result = {
"path": "src/auth/login.py",
"issues": [
{
"line": 47,
"severity": "CRITICAL",
"type": "SQL_INJECTION",
"message": "User input directly concatenated into SQL query",
"code_snippet": "query = f\"SELECT * FROM users WHERE username = '{username}'\"",
"fix": "Use parameterized queries: cursor.execute('SELECT * FROM users WHERE username = ?', (username,))",
},
{
"line": 89,
"severity": "HIGH",
"type": "HARDCODED_SECRET",
"message": "API key hardcoded in source code",
"code_snippet": 'API_KEY = "sk-prod-a1b2c3d4e5f6g7h8i9j0"',
"fix": "Use environment variables or secret management",
},
],
"metrics": {"complexity": 34, "coverage": 0.12},
}
elif i == 56:
# Another critical issue
file_result = {
"path": "src/api/upload.py",
"issues": [
{
"line": 112,
"severity": "CRITICAL",
"type": "PATH_TRAVERSAL",
"message": "File path not sanitized, allows directory traversal",
"code_snippet": "file_path = os.path.join(UPLOAD_DIR, user_filename)",
"fix": "Use secure_filename() and validate path stays within UPLOAD_DIR",
}
],
"metrics": {"complexity": 28, "coverage": 0.45},
}
else:
file_result = {
"path": f"src/module_{i}/handler.py",
"issues": [
{
"line": 10 + (i % 50),
"severity": "LOW",
"type": "STYLE",
"message": "Line too long (> 100 characters)",
}
]
if i % 4 == 0
else [],
"metrics": {"complexity": 5 + (i % 15), "coverage": 0.7 + (i % 30) / 100},
}
files.append(file_result)
messages = [
{"role": "user", "content": "Run a security scan on the codebase"},
{
"role": "assistant",
"content": [
{
"type": "tool_use",
"id": "call_scan",
"name": "security_scan",
"input": {"path": "src/"},
}
],
},
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "call_scan",
"content": json.dumps(files, indent=2),
}
],
},
{
"role": "user",
"content": "What are the critical security vulnerabilities I need to fix immediately?",
},
]
critical_facts = [
CriticalFact(
description="SQL injection file location",
value="src/auth/login.py line 47",
verification_question="In which file and line number is the SQL injection vulnerability?",
expected_answer_contains=["login.py", "47"],
),
CriticalFact(
description="Hardcoded API key value",
value="sk-prod-a1b2c3d4e5f6g7h8i9j0",
verification_question="What is the hardcoded API key that was found?",
expected_answer_contains=["sk-prod", "a1b2c3d4"],
),
CriticalFact(
description="Path traversal file",
value="src/api/upload.py",
verification_question="Which file has the path traversal vulnerability?",
expected_answer_contains=["upload.py"],
),
CriticalFact(
description="Path traversal line",
value="line 112",
verification_question="What line number has the path traversal issue in upload.py?",
expected_answer_contains=["112"],
),
]
return EvalTestCase(
name="code_review_needles",
description="80 file scan results with critical vulnerabilities at positions 23 and 56",
messages=messages,
critical_facts=critical_facts,
)
# =============================================================================
# EVALUATION ENGINE
# =============================================================================
class QualityRetentionEvaluator:
"""Evaluates whether compression retains critical information."""
# Tool definitions required by Anthropic API when messages contain tool results
TOOL_DEFINITIONS = [
{
"name": "search_incidents",
"description": "Search for incidents by date",
"input_schema": {
"type": "object",
"properties": {
"date": {"type": "string", "description": "Date to search (YYYY-MM-DD)"}
},
"required": ["date"],
},
},
{
"name": "get_logs",
"description": "Get logs for a time range",
"input_schema": {
"type": "object",
"properties": {
"timerange": {"type": "string", "description": "Time range (e.g., '1h', '24h')"}
},
"required": ["timerange"],
},
},
{
"name": "security_scan",
"description": "Run security scan on codebase",
"input_schema": {
"type": "object",
"properties": {"path": {"type": "string", "description": "Path to scan"}},
"required": ["path"],
},
},
]
def __init__(self, api_key: str):
self.api_key = api_key
self.base_client = Anthropic(api_key=api_key)
self.provider = AnthropicProvider()
self.tokenizer = Tokenizer(TiktokenCounter())
def _create_client(self, config: HeadroomConfig) -> HeadroomClient:
"""Create a HeadroomClient with the given config."""
return HeadroomClient(
original_client=Anthropic(api_key=self.api_key),
provider=self.provider,
default_mode="optimize",
config=config,
)
def _ask_question(
self,
client: HeadroomClient,
messages: list[dict],
question: str,
system_prompt: str,
) -> str:
"""Ask a verification question and get the response."""
# Add the verification question to the conversation
eval_messages = messages + [{"role": "user", "content": question}]
response = client.messages.create(
model="claude-sonnet-4-20250514",
messages=eval_messages,
system=system_prompt,
tools=self.TOOL_DEFINITIONS, # Required when messages contain tool results
max_tokens=300,
)
if hasattr(response, "content") and response.content:
if isinstance(response.content, list):
return response.content[0].text if response.content else ""
return str(response.content)
return str(response)
def _check_answer(self, answer: str, expected_contains: list[str]) -> bool:
"""Check if answer contains expected content."""
answer_lower = answer.lower()
for expected in expected_contains:
if expected.lower() in answer_lower:
return True
return False
def evaluate_test_case(
self,
test_case: EvalTestCase,
config: HeadroomConfig,
) -> EvalResult:
"""Evaluate a single test case."""
print(f"\n{'=' * 60}")
print(f"EVAL: {test_case.name}")
print(f"{'=' * 60}")
print(f"Description: {test_case.description}")
# Count tokens before compression
tokens_before = self.tokenizer.count_messages(test_case.messages)
print(f"Tokens before: {tokens_before:,}")
# Create client with compression enabled
client = self._create_client(config)
# Simulate compression to see what we get
try:
sim_result = client.messages.simulate(
model="claude-sonnet-4-20250514",
messages=test_case.messages,
system=test_case.system_prompt,
tools=self.TOOL_DEFINITIONS, # Required when messages contain tool results
)
tokens_after = sim_result.tokens_after
compression_ratio = 1 - (tokens_after / tokens_before) if tokens_before > 0 else 0
print(f"Tokens after: {tokens_after:,}")
print(f"Compression: {compression_ratio * 100:.1f}%")
print(f"Transforms: {sim_result.transforms[:3]}...") # First 3
except Exception as e:
return EvalResult(
test_name=test_case.name,
tokens_before=tokens_before,
tokens_after=0,
compression_ratio=0,
facts_tested=len(test_case.critical_facts),
facts_retained=0,
retention_rate=0,
quality_adjusted_savings=0,
error=str(e),
)
# Now verify each critical fact
print(f"\nVerifying {len(test_case.critical_facts)} critical facts...")
facts_retained = 0
details = []
for fact in test_case.critical_facts:
print(f"\n Fact: {fact.description}")
print(f" Question: {fact.verification_question}")
try:
# Ask the verification question (with compression applied)
answer = self._ask_question(
client,
test_case.messages,
fact.verification_question,
test_case.system_prompt,
)
# Check if answer contains expected content
retained = self._check_answer(answer, fact.expected_answer_contains)
if retained:
facts_retained += 1
print(" Result: ✅ RETAINED")
print(f" Answer: {answer[:100]}...")
else:
print(" Result: ❌ LOST")
print(f" Expected: {fact.expected_answer_contains}")
print(f" Got: {answer[:150]}...")
details.append(
{
"fact": fact.description,
"question": fact.verification_question,
"expected": fact.expected_answer_contains,
"answer": answer[:200],
"retained": retained,
}
)
except Exception as e:
print(f" Result: ❌ ERROR: {e}")
details.append(
{
"fact": fact.description,
"error": str(e),
"retained": False,
}
)
retention_rate = (
facts_retained / len(test_case.critical_facts) if test_case.critical_facts else 0
)
quality_adjusted_savings = compression_ratio * retention_rate
print(f"\n{'=' * 60}")
print(
f"RESULT: {facts_retained}/{len(test_case.critical_facts)} facts retained ({retention_rate * 100:.0f}%)"
)
print(f"Quality-Adjusted Savings: {quality_adjusted_savings * 100:.1f}%")
print(f"{'=' * 60}")
return EvalResult(
test_name=test_case.name,
tokens_before=tokens_before,
tokens_after=tokens_after,
compression_ratio=compression_ratio,
facts_tested=len(test_case.critical_facts),
facts_retained=facts_retained,
retention_rate=retention_rate,
quality_adjusted_savings=quality_adjusted_savings,
details=details,
)
def run_full_eval(self, config: HeadroomConfig) -> list[EvalResult]:
"""Run evaluation on all test cases."""
test_cases = [
create_search_results_with_needles(),
create_log_analysis_with_needles(),
create_code_review_with_needles(),
]
results = []
for test_case in test_cases:
result = self.evaluate_test_case(test_case, config)
results.append(result)
return results
# =============================================================================
# MAIN
# =============================================================================
def main():
print("\n" + "=" * 70)
print("QUALITY RETENTION EVALUATION")
print("=" * 70)
print("Verifying that intelligent compression retains critical information")
print("=" * 70)
# Create config with intelligent context enabled
config = HeadroomConfig()
config.intelligent_context = IntelligentContextConfig(
enabled=True,
use_importance_scoring=True,
compress_threshold=0.10,
summarize_threshold=0.25,
)
config.rolling_window.enabled = False
config.smart_crusher.enabled = True
# Run evaluation
evaluator = QualityRetentionEvaluator(API_KEY)
results = evaluator.run_full_eval(config)
# Summary
print("\n" + "=" * 70)
print("EVALUATION SUMMARY")
print("=" * 70)
total_facts = sum(r.facts_tested for r in results)
total_retained = sum(r.facts_retained for r in results)
total_tokens_before = sum(r.tokens_before for r in results)
total_tokens_after = sum(r.tokens_after for r in results)
print(f"\n{'Test Case':<30} {'Compression':<15} {'Retention':<15} {'Quality-Adj':<15}")
print("-" * 75)
for result in results:
status = (
"" if result.retention_rate >= 0.8 else "⚠️" if result.retention_rate >= 0.5 else ""
)
print(
f"{result.test_name:<30} "
f"{result.compression_ratio * 100:>6.1f}% "
f"{result.retention_rate * 100:>6.0f}% ({result.facts_retained}/{result.facts_tested}) "
f"{result.quality_adjusted_savings * 100:>6.1f}% {status}"
)
print("-" * 75)
overall_compression = (
1 - (total_tokens_after / total_tokens_before) if total_tokens_before > 0 else 0
)
overall_retention = total_retained / total_facts if total_facts > 0 else 0
overall_quality_adj = overall_compression * overall_retention
print(
f"{'OVERALL':<30} "
f"{overall_compression * 100:>6.1f}% "
f"{overall_retention * 100:>6.0f}% ({total_retained}/{total_facts}) "
f"{overall_quality_adj * 100:>6.1f}%"
)
print("\n" + "=" * 70)
if overall_retention >= 0.8:
print("✅ PASS: Critical information retention is good (>=80%)")
elif overall_retention >= 0.5:
print("⚠️ WARNING: Some critical information was lost (50-80% retention)")
else:
print("❌ FAIL: Significant critical information loss (<50% retention)")
print(
f"Tokens saved: {total_tokens_before - total_tokens_after:,} ({overall_compression * 100:.1f}% compression)"
)
print("=" * 70)
return 0 if overall_retention >= 0.8 else 1
if __name__ == "__main__":
sys.exit(main())

490
examples/real_data_demo.py Normal file
View file

@ -0,0 +1,490 @@
#!/usr/bin/env python3
"""
Real Data Demo: Headroom with Production-Scale Data
This demo uses realistic VOLUME of data to show Headroom's value:
- 100 log entries (one critical error buried inside)
- 50 pods (one unhealthy)
- 200 Prometheus metrics (a few critical ones)
- Real code, config, and service definitions
The scenario: Debug why a Kubernetes deployment is failing.
The agent must find the needle (database connection error) in the haystack.
Usage:
export ANTHROPIC_API_KEY=sk-ant-...
python examples/real_data_demo.py
"""
import json
import os
import time
from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.tools import tool
if not os.environ.get("ANTHROPIC_API_KEY"):
raise ValueError("ANTHROPIC_API_KEY environment variable required")
MODEL_ID = "claude-sonnet-4-20250514"
# =============================================================================
# REALISTIC DATA GENERATORS - Production-scale volume
# =============================================================================
def generate_app_logs(count: int = 100, error_at: int = 67) -> str:
"""Generate 100 log lines with critical error at position 67."""
lines = []
services = [
"api-gateway",
"user-service",
"order-service",
"payment-service",
"inventory-service",
]
for i in range(count):
ts = f"2024-01-15T14:{i // 60:02d}:{i % 60:02d}.{(i * 123) % 1000:03d}Z"
if i == error_at:
# THE CRITICAL ERROR - buried in the middle
lines.append(
f"{ts} ERROR [main] c.zaxxer.hikari.pool.HikariPool - HikariPool-1 - Exception during pool initialization"
)
lines.append(
"org.postgresql.util.PSQLException: Connection to db.internal:5432 refused. Check that the hostname and port are correct and that the postmaster is accepting TCP/IP connections."
)
lines.append(
" at org.postgresql.core.v3.ConnectionFactoryImpl.openConnectionImpl(ConnectionFactoryImpl.java:319)"
)
lines.append(
" at org.postgresql.core.ConnectionFactory.openConnection(ConnectionFactory.java:49)"
)
lines.append(" at org.postgresql.jdbc.PgConnection.<init>(PgConnection.java:223)")
lines.append(
" at com.zaxxer.hikari.util.DriverDataSource.getConnection(DriverDataSource.java:138)"
)
lines.append(" at com.zaxxer.hikari.pool.PoolBase.newConnection(PoolBase.java:364)")
lines.append("Caused by: java.net.ConnectException: Connection refused")
lines.append(" at java.base/sun.nio.ch.Net.pollConnect(Native Method)")
lines.append(" ... 15 more")
else:
svc = services[i % len(services)]
# Normal INFO logs - the haystack
messages = [
f"Request processed successfully - latency={50 + i % 30}ms",
f"Cache hit for key user:{i * 7}",
"Connection pool stats: active=5, idle=10, total=15",
"Health check passed - all dependencies healthy",
f"Processed batch of {10 + i % 20} items in {100 + i % 50}ms",
]
lines.append(
f"{ts} INFO [{svc}] c.m.a.{svc.replace('-', '.')} - {messages[i % len(messages)]}"
)
return "\n".join(lines)
def generate_pod_list(count: int = 50, unhealthy_at: int = 23) -> str:
"""Generate kubectl get pods output with one unhealthy pod."""
header = (
"NAME READY STATUS RESTARTS AGE"
)
lines = [header]
deployments = [
"api-server",
"user-service",
"order-service",
"payment-service",
"inventory-service",
"cache-service",
]
for i in range(count):
deploy = deployments[i % len(deployments)]
suffix = f"{i:05x}"[:5]
name = f"{deploy}-{suffix[:5]}-{suffix[2:]}"
if i == unhealthy_at:
# THE UNHEALTHY POD
lines.append(
"api-server-7f8d9-m4n2p 0/1 CrashLoopBackOff 5 (30s ago) 10m"
)
else:
# Healthy pods - the haystack
age = f"{(i * 3) % 24}h" if i % 3 == 0 else f"{(i * 7) % 30}d"
lines.append(f"{name:<42} 1/1 Running 0 {age}")
return "\n".join(lines)
def generate_prometheus_metrics(count: int = 200, critical_at: list = None) -> str:
"""Generate Prometheus metrics response with critical metrics mixed in."""
if critical_at is None:
critical_at = [47, 123, 156] # Positions of critical metrics
results = []
# Common metric types
metric_types = [
("http_requests_total", ["handler", "method", "status"]),
("http_request_duration_seconds", ["handler", "method"]),
("process_cpu_seconds_total", ["instance"]),
("go_goroutines", ["instance"]),
("node_memory_MemAvailable_bytes", ["instance"]),
("container_memory_usage_bytes", ["pod", "container"]),
]
handlers = ["/api/users", "/api/orders", "/api/products", "/api/health", "/api/metrics"]
instances = ["10.244.1.10:8080", "10.244.1.11:8080", "10.244.2.15:8080", "10.244.3.20:8080"]
pods = [
"api-server-abc12",
"user-service-def34",
"order-service-ghi56",
"payment-service-jkl78",
]
for i in range(count):
ts = 1705329825 + i
if i in critical_at:
# Critical metrics showing the problem
if i == critical_at[0]:
results.append(
{
"metric": {
"__name__": "up",
"job": "postgresql",
"instance": "db.internal:5432",
},
"value": [ts, "0"], # DATABASE IS DOWN
}
)
elif i == critical_at[1]:
results.append(
{
"metric": {"__name__": "pg_up", "datname": "myapp_production"},
"value": [ts, "0"], # DATABASE NOT ACCEPTING CONNECTIONS
}
)
elif i == critical_at[2]:
results.append(
{
"metric": {
"__name__": "kube_pod_container_status_restarts_total",
"pod": "api-server-7f8d9-m4n2p",
},
"value": [ts, "5"], # POD RESTARTING
}
)
else:
# Normal healthy metrics - the haystack
metric_name, label_keys = metric_types[i % len(metric_types)]
metric = {"__name__": metric_name}
if "handler" in label_keys:
metric["handler"] = handlers[i % len(handlers)]
metric["method"] = "GET" if i % 2 == 0 else "POST"
if "status" in label_keys:
metric["status"] = "200" # All healthy
if "instance" in label_keys:
metric["instance"] = instances[i % len(instances)]
if "pod" in label_keys:
metric["pod"] = pods[i % len(pods)]
metric["container"] = "main"
# Normal values
value = str(1000 + (i * 17) % 5000)
if "duration" in metric_name:
value = f"0.{(50 + i % 200):03d}"
elif metric_name == "up":
value = "1" # Healthy
results.append({"metric": metric, "value": [ts, value]})
return json.dumps(
{"status": "success", "data": {"resultType": "vector", "result": results}}, indent=2
)
# Generate the production-scale data
APP_LOGS = generate_app_logs(100, error_at=67)
POD_LIST = generate_pod_list(50, unhealthy_at=23)
PROMETHEUS_METRICS = generate_prometheus_metrics(200)
# Static realistic content (not repetitive, but real)
K8S_POD_DESCRIBE = """Name: api-server-7f8d9-m4n2p
Namespace: production
Status: Running
IP: 10.244.2.45
Containers:
api-server:
State: Waiting
Reason: CrashLoopBackOff
Ready: False
Restart Count: 5
Limits:
cpu: 500m
memory: 512Mi
Environment:
DATABASE_URL: <set to the key 'url' in secret 'db-credentials'>
Events:
Type Reason Age Message
---- ------ ---- -------
Warning Unhealthy 4m (x3 over 4m) Readiness probe failed: HTTP 503
Warning Unhealthy 3m (x9 over 4m) Liveness probe failed: HTTP 503
Warning BackOff 30s (x5 over 2m) Back-off restarting failed container"""
K8S_SERVICES = """NAME TYPE CLUSTER-IP EXTERNAL-IP PORT(S) AGE
api-server ClusterIP 10.96.45.123 <none> 8080/TCP 30d
db ClusterIP 10.96.78.234 <none> 5432/TCP 30d
redis-master ClusterIP 10.96.12.345 <none> 6379/TCP 30d
ENDPOINTS:
NAME ENDPOINTS AGE
api-server 10.244.2.45:8080,10.244.1.33:8080 30d
db <none> 30d
redis-master 10.244.3.67:6379 30d"""
HEALTH_CHECK_CODE = '''"""Health check endpoints for Kubernetes probes."""
from fastapi import APIRouter, Response, status
from sqlalchemy import text
router = APIRouter()
@router.get("/health")
async def health_check(response: Response):
"""Liveness probe - checks database connectivity."""
try:
async with engine.connect() as conn:
await conn.execute(text("SELECT 1"))
return {"status": "healthy"}
except Exception as e:
response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE
return {"status": "unhealthy", "error": str(e)}
'''
DB_CONFIG = """# config/database.yaml
production:
adapter: postgresql
host: db.internal
port: 5432
database: myapp_production
pool: 25
timeout: 5000
max_connections: 100"""
# =============================================================================
# TOOLS - Return production-scale data
# =============================================================================
@tool(name="kubectl_describe_pod")
def kubectl_describe_pod(pod_name: str) -> str:
"""Describe a Kubernetes pod."""
return K8S_POD_DESCRIBE
@tool(name="kubectl_get_pods")
def kubectl_get_pods(namespace: str = "production") -> str:
"""List all pods in namespace."""
return POD_LIST
@tool(name="kubectl_get_services")
def kubectl_get_services(namespace: str = "production") -> str:
"""Get Kubernetes services and endpoints."""
return K8S_SERVICES
@tool(name="get_application_logs")
def get_application_logs(pod_name: str, lines: int = 100) -> str:
"""Get application logs from a pod."""
return APP_LOGS
@tool(name="get_source_code")
def get_source_code(file_path: str) -> str:
"""Read source code file."""
return HEALTH_CHECK_CODE
@tool(name="get_config_file")
def get_config_file(path: str) -> str:
"""Read a configuration file."""
return DB_CONFIG
@tool(name="query_prometheus")
def query_prometheus(query: str) -> str:
"""Query Prometheus metrics."""
return PROMETHEUS_METRICS
# =============================================================================
# GROUND TRUTH - What we expect the agent to find
# =============================================================================
def verify_response(response: str) -> dict[str, bool]:
"""Verify the agent found the key information."""
response_lower = response.lower()
return {
"found_db_error": any(
term in response_lower
for term in ["connection refused", "db.internal", "5432", "postgresql", "psqlexception"]
),
"found_pod_issue": any(
term in response_lower
for term in ["crashloop", "restart", "unhealthy", "503", "probe failed"]
),
"found_endpoint_issue": any(
term in response_lower for term in ["endpoint", "no endpoints", "<none>", "db service"]
),
"found_root_cause": any(
term in response_lower for term in ["database", "connection", "postgres"]
)
and "down" in response_lower
or "fail" in response_lower
or "refused" in response_lower,
}
# =============================================================================
# MAIN DEMO
# =============================================================================
def main():
from headroom.integrations.agno import HeadroomAgnoModel
from headroom.pricing import estimate_cost
print("\n" + "=" * 70)
print(" REAL DATA DEMO: Production-Scale Kubernetes Investigation")
print("=" * 70)
# Show what data we're using
data_sources = {
"Application logs (100 entries)": APP_LOGS,
"Pod list (50 pods)": POD_LIST,
"Prometheus metrics (200 series)": PROMETHEUS_METRICS,
"Pod describe": K8S_POD_DESCRIBE,
"Services/Endpoints": K8S_SERVICES,
"Health check code": HEALTH_CHECK_CODE,
"Database config": DB_CONFIG,
}
total_chars = sum(len(v) for v in data_sources.values())
print("\n Data sources (production-scale volume):")
for name, data in data_sources.items():
print(f" {name:<35} {len(data):>8,} chars")
print(f" {'' * 50}")
print(f" {'TOTAL':<35} {total_chars:>8,} chars")
print("\n Content types: Java stack traces, K8s YAML, Python code, Prometheus JSON")
print(" Challenge: Find critical error at position 67 in 100 log entries")
# Create agent with Headroom
base_model = Claude(id=MODEL_ID)
model = HeadroomAgnoModel(wrapped_model=base_model)
agent = Agent(
model=model,
tools=[
kubectl_describe_pod,
kubectl_get_pods,
kubectl_get_services,
get_application_logs,
get_source_code,
get_config_file,
query_prometheus,
],
markdown=True,
)
question = """Our api-server deployment in production is failing. Pods keep restarting.
Please investigate:
1. List all pods and identify unhealthy ones
2. Describe the problematic pod
3. Check application logs for errors
4. Verify services and endpoints
5. Check Prometheus metrics for anomalies
Find the ROOT CAUSE and explain what's failing."""
print("\n Running investigation...")
start = time.time()
response = agent.run(question)
response_text = response.content if hasattr(response, "content") else str(response)
duration = time.time() - start
# Get Headroom stats
stats = model.get_savings_summary()
tokens_before = stats["total_tokens_before"]
tokens_after = stats["total_tokens_after"]
tokens_saved = stats["total_tokens_saved"]
pct_saved = (tokens_saved / tokens_before * 100) if tokens_before > 0 else 0
# Calculate costs
cost_before = estimate_cost(MODEL_ID, input_tokens=tokens_before)
cost_after = estimate_cost(MODEL_ID, input_tokens=tokens_after)
# Verify findings
verification = verify_response(response_text)
findings_found = sum(verification.values())
# Results
print("\n" + "=" * 70)
print(" RESULTS")
print("=" * 70)
print(f"""
Without With
Headroom Headroom Savings
Input Tokens {tokens_before:>8,} {tokens_after:>8,} {tokens_saved:,} ({pct_saved:.0f}%)""")
if cost_before and cost_after:
cost_saved = cost_before - cost_after
print(
f" Input Cost ${cost_before:.4f} ${cost_after:.4f} ${cost_saved:.4f}"
)
print(f"""
Duration {duration:.1f}s
Tool calls {stats["total_requests"]}
Ground Truth Verification ({findings_found}/4 findings):""")
for name, found in verification.items():
print(f" {'' if found else ''} {name.replace('_', ' ').title()}")
print("\n" + "=" * 70)
print(" AGENT RESPONSE (excerpt)")
print("=" * 70)
# Show first 2000 chars of response
excerpt = response_text[:2000] + "..." if len(response_text) > 2000 else response_text
print(excerpt)
print("\n" + "=" * 70)
if findings_found >= 3 and pct_saved > 30:
print(f" SUCCESS: {pct_saved:.0f}% compression with {findings_found}/4 findings preserved")
elif findings_found >= 3:
print(f" ACCURACY OK: {findings_found}/4 findings, but only {pct_saved:.0f}% compression")
else:
print(f" WARNING: Only {findings_found}/4 findings - compression may be too aggressive")
print("=" * 70 + "\n")
if __name__ == "__main__":
main()

View file

@ -0,0 +1,385 @@
#!/usr/bin/env python3
"""
Comprehensive Multi-Agent Reasoning Test with Debugging
This test creates:
1. A reasoning agent (reasoning=True)
2. Multiple tool-using agents
3. Tests message flow inter and intra agent
We run WITHOUT Headroom first, then WITH Headroom to find where the issue occurs.
"""
import json
import os
import sys
import traceback
from typing import Any
# Enable maximum Agno debugging
os.environ["AGNO_DEBUG"] = "true"
from agno.agent import Agent
from agno.models.anthropic import Claude
from agno.tools import tool
# Check for API key
API_KEY = os.environ.get("ANTHROPIC_API_KEY")
if not API_KEY:
print("ERROR: ANTHROPIC_API_KEY environment variable required")
sys.exit(1)
# =============================================================================
# DEBUGGING UTILITIES
# =============================================================================
DEBUG_LOG = []
def debug_log(category: str, message: str, data: Any = None):
"""Log debug information."""
entry = f"[{category}] {message}"
if data is not None:
if isinstance(data, list):
entry += f"\n Items: {len(data)}"
for i, item in enumerate(data[:5]): # First 5 items
item_type = type(item).__name__
has_log = hasattr(item, "log")
has_content = hasattr(item, "content")
if isinstance(item, dict):
keys = list(item.keys())
entry += f"\n [{i}] dict with keys: {keys}"
else:
entry += (
f"\n [{i}] {item_type} (has .log={has_log}, has .content={has_content})"
)
else:
entry += f"\n Data: {type(data).__name__}"
DEBUG_LOG.append(entry)
print(entry)
def dump_message_details(messages: list, label: str):
"""Dump detailed message information."""
print(f"\n{'=' * 60}")
print(f"MESSAGE DUMP: {label}")
print(f"{'=' * 60}")
print(f"Total messages: {len(messages)}")
for i, msg in enumerate(messages):
print(f"\n--- Message {i} ---")
print(f" Type: {type(msg).__name__}")
print(f" Is dict: {isinstance(msg, dict)}")
print(f" Has .log(): {hasattr(msg, 'log')}")
print(f" Has .content: {hasattr(msg, 'content')}")
if isinstance(msg, dict):
print(f" Keys: {list(msg.keys())}")
print(f" Role: {msg.get('role', 'N/A')}")
content = msg.get("content", "")
print(f" Content preview: {str(content)[:100]}...")
elif hasattr(msg, "role"):
print(f" Role: {msg.role}")
content = getattr(msg, "content", "")
print(f" Content preview: {str(content)[:100]}...")
# Try calling .log() to see if it works
if hasattr(msg, "log"):
try:
msg.log(metrics=False)
print(" .log() call: SUCCESS")
except Exception as e:
print(f" .log() call: FAILED - {e}")
print(f"{'=' * 60}\n")
# =============================================================================
# MOCK TOOLS
# =============================================================================
@tool(name="search_knowledge_base")
def search_knowledge_base(query: str) -> str:
"""Search the knowledge base for information.
Args:
query: Search query
Returns:
Search results as JSON
"""
debug_log("TOOL", f"search_knowledge_base called with: {query}")
results = [
{
"id": 1,
"title": "Memory Management Best Practices",
"content": "Always release resources...",
},
{"id": 2, "title": "Worker Pool Optimization", "content": "Use thread pool executors..."},
{"id": 3, "title": "Garbage Collection Tuning", "content": "Set appropriate heap sizes..."},
]
return json.dumps(results, indent=2)
@tool(name="analyze_code")
def analyze_code(file_path: str) -> str:
"""Analyze code for issues.
Args:
file_path: Path to the file to analyze
Returns:
Analysis results as JSON
"""
debug_log("TOOL", f"analyze_code called with: {file_path}")
return json.dumps(
{
"file": file_path,
"issues": [
{"line": 42, "type": "memory_leak", "description": "Resource not released"},
{"line": 87, "type": "performance", "description": "Inefficient loop"},
],
"suggestions": ["Add cleanup in finally block", "Use list comprehension"],
},
indent=2,
)
# =============================================================================
# TEST FUNCTIONS
# =============================================================================
def test_simple_agent(use_headroom: bool, use_reasoning: bool) -> dict:
"""Test a simple agent configuration."""
label = f"{'WITH' if use_headroom else 'WITHOUT'} Headroom, reasoning={use_reasoning}"
print(f"\n{'#' * 70}")
print(f"# TEST: {label}")
print(f"{'#' * 70}")
DEBUG_LOG.clear()
try:
# Create the model
if use_headroom:
from headroom.integrations.agno import HeadroomAgnoModel
base_model = Claude(id="claude-sonnet-4-20250514")
model = HeadroomAgnoModel(wrapped_model=base_model)
debug_log("SETUP", "Created HeadroomAgnoModel wrapping Claude")
else:
model = Claude(id="claude-sonnet-4-20250514")
debug_log("SETUP", "Created Claude model directly")
# Create the agent
agent = Agent(
model=model,
tools=[search_knowledge_base, analyze_code],
reasoning=use_reasoning,
markdown=True,
debug_mode=True,
)
debug_log("SETUP", f"Created Agent with reasoning={use_reasoning}")
# Simple question that uses tools
question = "Search the knowledge base for memory management and analyze worker.py for issues. Summarize what you find."
debug_log("INPUT", f"Question: {question}")
# Run the agent
debug_log("RUN", "Starting agent.run()...")
response = agent.run(question)
# Extract response
if hasattr(response, "content") and response.content is not None:
response_text = response.content
elif response is not None:
response_text = str(response)
else:
response_text = "(No response content)"
debug_log("OUTPUT", f"Response length: {len(response_text)} chars")
debug_log("OUTPUT", f"Response preview: {response_text[:200]}...")
# Get Headroom stats if available
headroom_stats = None
if use_headroom and hasattr(model, "get_savings_summary"):
headroom_stats = model.get_savings_summary()
debug_log("HEADROOM", f"Stats: {headroom_stats}")
return {
"success": True,
"label": label,
"response_length": len(response_text),
"response_preview": response_text[:500],
"headroom_stats": headroom_stats,
"debug_log": DEBUG_LOG.copy(),
}
except Exception as e:
error_msg = str(e)
tb = traceback.format_exc()
debug_log("ERROR", f"Exception: {error_msg}")
debug_log("ERROR", f"Traceback:\n{tb}")
return {
"success": False,
"label": label,
"error": error_msg,
"traceback": tb,
"debug_log": DEBUG_LOG.copy(),
}
def test_with_message_interception(use_headroom: bool, use_reasoning: bool) -> dict:
"""Test with message interception to see what's being passed around."""
label = (
f"INTERCEPTED: {'WITH' if use_headroom else 'WITHOUT'} Headroom, reasoning={use_reasoning}"
)
print(f"\n{'#' * 70}")
print(f"# TEST: {label}")
print(f"{'#' * 70}")
DEBUG_LOG.clear()
# Patch Agno's _log_messages to intercept and debug
original_log_messages = None
try:
from agno.models import base as agno_base
original_log_messages = agno_base._log_messages
def intercepted_log_messages(messages):
debug_log("INTERCEPT", "_log_messages called", messages)
dump_message_details(messages, "_log_messages input")
# Check each message
for i, msg in enumerate(messages):
if isinstance(msg, dict):
debug_log("INTERCEPT", f"Message {i} is a DICT - this will fail!")
elif not hasattr(msg, "log"):
debug_log("INTERCEPT", f"Message {i} has no .log() method!")
# Call original
return original_log_messages(messages)
agno_base._log_messages = intercepted_log_messages
debug_log("SETUP", "Patched _log_messages for interception")
# Now run the actual test
result = test_simple_agent(use_headroom, use_reasoning)
result["label"] = label
return result
except Exception as e:
error_msg = str(e)
tb = traceback.format_exc()
debug_log("ERROR", f"Exception: {error_msg}")
debug_log("ERROR", f"Traceback:\n{tb}")
return {
"success": False,
"label": label,
"error": error_msg,
"traceback": tb,
"debug_log": DEBUG_LOG.copy(),
}
finally:
# Restore original
if original_log_messages:
agno_base._log_messages = original_log_messages
debug_log("CLEANUP", "Restored original _log_messages")
def run_all_tests():
"""Run all test combinations."""
print("\n" + "=" * 70)
print("COMPREHENSIVE MULTI-AGENT REASONING TEST")
print("=" * 70)
print(f"API Key: {'SET' if API_KEY else 'NOT SET'}")
print("=" * 70)
results = []
# Test matrix
test_cases = [
# (use_headroom, use_reasoning, use_interception)
(False, False, False), # Baseline: No Headroom, No Reasoning
(False, True, False), # No Headroom, With Reasoning
(True, False, False), # With Headroom, No Reasoning
(True, True, False), # With Headroom, With Reasoning
(True, True, True), # With Headroom, With Reasoning, With Interception
]
for use_headroom, use_reasoning, use_interception in test_cases:
print(f"\n{'=' * 70}")
print(
f"Running: Headroom={use_headroom}, Reasoning={use_reasoning}, Intercept={use_interception}"
)
print("=" * 70)
try:
if use_interception:
result = test_with_message_interception(use_headroom, use_reasoning)
else:
result = test_simple_agent(use_headroom, use_reasoning)
results.append(result)
except Exception as e:
results.append(
{
"success": False,
"label": f"Headroom={use_headroom}, Reasoning={use_reasoning}",
"error": str(e),
"traceback": traceback.format_exc(),
}
)
print(f"\nResult: {'SUCCESS' if results[-1]['success'] else 'FAILED'}")
if not results[-1]["success"]:
print(f"Error: {results[-1].get('error', 'Unknown')}")
# Summary
print("\n" + "=" * 70)
print("SUMMARY")
print("=" * 70)
for result in results:
status = "✅ PASS" if result["success"] else "❌ FAIL"
print(f"{status} - {result['label']}")
if not result["success"]:
print(f" Error: {result.get('error', 'Unknown')[:100]}")
elif result.get("headroom_stats"):
stats = result["headroom_stats"]
saved = stats.get("total_tokens_saved", 0)
before = stats.get("total_tokens_before", 0)
pct = (saved / before * 100) if before > 0 else 0
print(f" Tokens saved: {saved:,} ({pct:.1f}%)")
print("\n" + "=" * 70)
# Detailed failure analysis
failures = [r for r in results if not r["success"]]
if failures:
print("\nDETAILED FAILURE ANALYSIS")
print("=" * 70)
for failure in failures:
print(f"\n--- {failure['label']} ---")
print(f"Error: {failure.get('error', 'Unknown')}")
if "traceback" in failure:
print(f"Traceback:\n{failure['traceback']}")
if "debug_log" in failure:
print("\nDebug Log:")
for entry in failure["debug_log"][-20:]: # Last 20 entries
print(f" {entry}")
return results
if __name__ == "__main__":
run_all_tests()

View file

@ -266,6 +266,7 @@ class HeadroomClient:
cache_optimizer: BaseCacheOptimizer | None = None,
enable_cache_optimizer: bool = True,
enable_semantic_cache: bool = False,
config: HeadroomConfig | None = None,
):
"""
Initialize HeadroomClient.
@ -280,6 +281,9 @@ class HeadroomClient:
enable_cache_optimizer=True, auto-detects from provider.
enable_cache_optimizer: Enable provider-specific cache optimization.
enable_semantic_cache: Enable query-level semantic caching.
config: Optional HeadroomConfig for full control over all settings
including intelligent_context. When provided, takes precedence
over individual settings like store_url, default_mode, etc.
"""
self._original = original_client
self._provider = provider
@ -295,12 +299,22 @@ class HeadroomClient:
self._store_url = store_url
self._default_mode = HeadroomMode(default_mode)
# Build config
self._config = HeadroomConfig()
self._config.store_url = store_url
self._config.default_mode = self._default_mode
self._config.cache_optimizer.enabled = enable_cache_optimizer
self._config.cache_optimizer.enable_semantic_cache = enable_semantic_cache
# Use provided config or build from individual parameters
if config is not None:
self._config = config
# Override store_url and mode if explicitly provided in config
if config.store_url:
self._store_url = config.store_url
else:
self._config.store_url = store_url
self._default_mode = config.default_mode
else:
# Build config from individual parameters
self._config = HeadroomConfig()
self._config.store_url = store_url
self._config.default_mode = self._default_mode
self._config.cache_optimizer.enabled = enable_cache_optimizer
self._config.cache_optimizer.enable_semantic_cache = enable_semantic_cache
if model_context_limits:
self._config.model_context_limits.update(model_context_limits)

View file

@ -384,6 +384,12 @@ class HeadroomConfig:
cache_optimizer: CacheOptimizerConfig = field(default_factory=CacheOptimizerConfig)
ccr: CCRConfig = field(default_factory=CCRConfig) # Compress-Cache-Retrieve
# Intelligent context management (Phase 2.5)
# When enabled, replaces RollingWindow with semantic-aware context management
intelligent_context: IntelligentContextConfig = field(
default_factory=lambda: IntelligentContextConfig(enabled=False)
)
# Debugging - opt-in diff artifact generation
generate_diff_artifact: bool = False # Enable to get detailed transform diffs

View file

@ -2567,7 +2567,9 @@ if __name__ == "__main__":
# Server
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--port", type=int, default=8787)
parser.add_argument("--openai-api-url", help=f"Custom OpenAI API URL (default: {HeadroomProxy.OPENAI_API_URL})")
parser.add_argument(
"--openai-api-url", help=f"Custom OpenAI API URL (default: {HeadroomProxy.OPENAI_API_URL})"
)
# Optimization
parser.add_argument("--no-optimize", action="store_true", help="Disable optimization")

View file

@ -9,6 +9,7 @@ from ..config import (
CacheAlignerConfig,
DiffArtifact,
HeadroomConfig,
IntelligentContextConfig,
RollingWindowConfig,
ToolCrusherConfig,
TransformDiff,
@ -18,6 +19,7 @@ from ..tokenizer import Tokenizer
from ..utils import deep_copy_messages
from .base import Transform
from .cache_aligner import CacheAligner
from .intelligent_context import IntelligentContextManager
from .rolling_window import RollingWindow
from .smart_crusher import SmartCrusher
from .tool_crusher import ToolCrusher
@ -93,8 +95,17 @@ class TransformPipeline:
# Fallback to fixed-rule crushing
transforms.append(ToolCrusher(self.config.tool_crusher))
# 3. Rolling Window (enforce limits last)
if self.config.rolling_window.enabled:
# 3. Context Management (enforce limits last)
# IntelligentContextManager takes precedence over RollingWindow when enabled
if self.config.intelligent_context.enabled:
# Use semantic-aware context management with scoring
transforms.append(IntelligentContextManager(self.config.intelligent_context))
logger.info(
"Pipeline using IntelligentContextManager with strategies: "
"COMPRESS_FIRST -> SUMMARIZE -> DROP_BY_SCORE"
)
elif self.config.rolling_window.enabled:
# Fallback to position-based rolling window
transforms.append(RollingWindow(self.config.rolling_window))
return transforms
@ -273,6 +284,7 @@ def create_pipeline(
tool_crusher_config: ToolCrusherConfig | None = None,
cache_aligner_config: CacheAlignerConfig | None = None,
rolling_window_config: RollingWindowConfig | None = None,
intelligent_context_config: IntelligentContextConfig | None = None,
) -> TransformPipeline:
"""
Create a pipeline with specific configurations.
@ -281,6 +293,9 @@ def create_pipeline(
tool_crusher_config: Tool crusher configuration.
cache_aligner_config: Cache aligner configuration.
rolling_window_config: Rolling window configuration.
intelligent_context_config: Intelligent context configuration.
When provided with enabled=True, replaces RollingWindow with
semantic-aware context management.
Returns:
Configured TransformPipeline.
@ -293,5 +308,7 @@ def create_pipeline(
config.cache_aligner = cache_aligner_config
if rolling_window_config is not None:
config.rolling_window = rolling_window_config
if intelligent_context_config is not None:
config.intelligent_context = intelligent_context_config
return TransformPipeline(config)

View file

@ -0,0 +1,762 @@
"""Comprehensive tests for progressive summarization.
These tests verify that ProgressiveSummarizer works correctly with:
- Anchored summaries that track message positions
- Callback pattern for summarization (no internal LLM calls)
- CCR integration for retrieval
- Extractive fallback summarization
CRITICAL: NO MOCKS for core logic. All tests use real implementations.
"""
from __future__ import annotations
from typing import Any
import pytest
from headroom.tokenizer import Tokenizer
from headroom.tokenizers import EstimatingTokenCounter
from headroom.transforms.progressive_summarizer import (
AnchoredSummary,
ProgressiveSummarizer,
SummarizationResult,
extractive_summarizer,
)
# =============================================================================
# Test Fixtures
# =============================================================================
@pytest.fixture
def tokenizer() -> Tokenizer:
"""Create a tokenizer for testing."""
return Tokenizer(EstimatingTokenCounter())
@pytest.fixture
def simple_conversation() -> list[dict[str, Any]]:
"""Simple conversation without tool calls."""
return [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Hello, how are you?"},
{"role": "assistant", "content": "I'm doing well, thank you for asking!"},
{"role": "user", "content": "Can you help me with Python?"},
{"role": "assistant", "content": "Of course! What would you like to know?"},
{"role": "user", "content": "How do I read a file?"},
{
"role": "assistant",
"content": "You can use open() to read files. Here's an example: with open('file.txt', 'r') as f: content = f.read()",
},
]
@pytest.fixture
def conversation_with_tools() -> list[dict[str, Any]]:
"""Conversation with tool calls and responses."""
return [
{"role": "system", "content": "You are a helpful assistant with tools."},
{"role": "user", "content": "Search for information about Python."},
{
"role": "assistant",
"content": "I'll search for that.",
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {"name": "search", "arguments": "{}"},
}
],
},
{
"role": "tool",
"tool_call_id": "call_1",
"content": '{"results": [{"title": "Python Guide", "url": "example.com"}, {"title": "Python Tutorial", "url": "tutorial.com"}]}',
},
{"role": "assistant", "content": "Here's what I found about Python programming."},
{"role": "user", "content": "Thanks! Can you search for more?"},
{
"role": "assistant",
"content": "Sure, searching again for more results.",
"tool_calls": [
{
"id": "call_2",
"type": "function",
"function": {"name": "search", "arguments": "{}"},
}
],
},
{
"role": "tool",
"tool_call_id": "call_2",
"content": '{"results": [{"title": "Advanced Python", "status": "found"}, {"error": "Some results failed to load"}]}',
},
{"role": "assistant", "content": "Here are more results for you."},
]
@pytest.fixture
def long_conversation() -> list[dict[str, Any]]:
"""Long conversation for testing summarization scenarios."""
messages = [{"role": "system", "content": "You are a helpful assistant."}]
# Add many turns
for i in range(20):
messages.append(
{"role": "user", "content": f"This is question number {i}. What about topic {i}?"}
)
messages.append(
{
"role": "assistant",
"content": f"Here's my detailed response about topic {i}. " * 10
+ f"In summary, topic {i} is interesting.",
}
)
return messages
# =============================================================================
# AnchoredSummary Tests
# =============================================================================
class TestAnchoredSummary:
"""Tests for AnchoredSummary dataclass."""
def test_compression_ratio_calculation(self) -> None:
"""Test compression ratio is calculated correctly."""
summary = AnchoredSummary(
summary_text="Summary",
start_index=0,
end_index=5,
original_message_count=6,
original_tokens=1000,
summary_tokens=100,
)
assert summary.compression_ratio == 0.1 # 100/1000
def test_compression_ratio_with_zero_original(self) -> None:
"""Test compression ratio handles zero original tokens."""
summary = AnchoredSummary(
summary_text="Summary",
start_index=0,
end_index=0,
original_message_count=1,
original_tokens=0,
summary_tokens=10,
)
assert summary.compression_ratio == 1.0 # fallback
def test_tokens_saved(self) -> None:
"""Test tokens_saved calculation."""
summary = AnchoredSummary(
summary_text="Summary",
start_index=0,
end_index=5,
original_message_count=6,
original_tokens=1000,
summary_tokens=100,
)
assert summary.tokens_saved == 900
def test_tokens_saved_no_negative(self) -> None:
"""Test tokens_saved doesn't go negative."""
summary = AnchoredSummary(
summary_text="Long summary that is bigger than original",
start_index=0,
end_index=0,
original_message_count=1,
original_tokens=10,
summary_tokens=50,
)
assert summary.tokens_saved == 0 # max(0, ...)
def test_optional_fields(self) -> None:
"""Test optional fields have defaults."""
summary = AnchoredSummary(
summary_text="Summary",
start_index=0,
end_index=5,
original_message_count=6,
original_tokens=1000,
summary_tokens=100,
)
assert summary.cache_hash is None
assert summary.tool_names == []
assert summary.created_at > 0
# =============================================================================
# Extractive Summarizer Tests
# =============================================================================
class TestExtractiveSummarizer:
"""Tests for the default extractive summarizer."""
def test_empty_messages(self) -> None:
"""Test handling of empty message list."""
result = extractive_summarizer([])
assert result == "[No messages to summarize]"
def test_simple_conversation(self, simple_conversation: list[dict[str, Any]]) -> None:
"""Test summarization of simple conversation."""
# Skip system message, use rest
result = extractive_summarizer(simple_conversation[1:])
assert "[Summary of 6 messages]" in result
assert "user messages" in result
assert "assistant" in result.lower()
def test_tool_messages_detection(self, conversation_with_tools: list[dict[str, Any]]) -> None:
"""Test that tool messages are detected and counted."""
result = extractive_summarizer(conversation_with_tools)
assert "tool outputs" in result.lower()
def test_error_detection_in_tools(self) -> None:
"""Test that errors in tool responses are detected."""
messages = [
{
"role": "tool",
"tool_call_id": "call_1",
"content": "Error: Connection failed",
},
]
result = extractive_summarizer(messages)
assert "with errors" in result
def test_successful_tools(self) -> None:
"""Test that successful tool responses are marked correctly."""
messages = [
{
"role": "tool",
"tool_call_id": "call_1",
"content": '{"status": "success", "data": [1, 2, 3]}',
},
]
result = extractive_summarizer(messages)
assert "successful" in result
def test_long_assistant_content_truncated(self) -> None:
"""Test that long assistant content is truncated."""
messages = [
{"role": "assistant", "content": "X" * 200},
]
result = extractive_summarizer(messages)
assert "..." in result # Truncation indicator
def test_context_ignored(self) -> None:
"""Test that context parameter exists but doesn't change output format."""
messages = [{"role": "user", "content": "Hello"}]
result1 = extractive_summarizer(messages, context="")
result2 = extractive_summarizer(messages, context="Some context here")
# Both should work (context is unused in extractive mode)
assert "[Summary of 1 messages]" in result1
assert "[Summary of 1 messages]" in result2
# =============================================================================
# ProgressiveSummarizer Core Tests
# =============================================================================
class TestProgressiveSummarizerInit:
"""Tests for ProgressiveSummarizer initialization."""
def test_default_init(self) -> None:
"""Test default initialization."""
summarizer = ProgressiveSummarizer()
assert summarizer.max_summary_tokens == 500
assert summarizer.min_messages_to_summarize == 3
assert summarizer.store_for_retrieval is True
# Default summarizer is extractive_summarizer
assert summarizer.summarize_fn is not None
def test_custom_summarize_fn(self) -> None:
"""Test custom summarization function."""
def custom_fn(messages: list[dict], context: str = "") -> str:
return f"Custom: {len(messages)} messages"
summarizer = ProgressiveSummarizer(summarize_fn=custom_fn)
result = summarizer.summarize_fn([{"role": "user", "content": "test"}])
assert "Custom: 1" in result
def test_custom_config(self) -> None:
"""Test custom configuration."""
summarizer = ProgressiveSummarizer(
max_summary_tokens=1000,
min_messages_to_summarize=5,
store_for_retrieval=False,
)
assert summarizer.max_summary_tokens == 1000
assert summarizer.min_messages_to_summarize == 5
assert summarizer.store_for_retrieval is False
# =============================================================================
# Find Candidates Tests
# =============================================================================
class TestFindSummarizationCandidates:
"""Tests for finding candidate message groups."""
def test_no_protected_all_candidates(self) -> None:
"""All messages are candidates when none protected."""
summarizer = ProgressiveSummarizer(min_messages_to_summarize=3)
messages = [
{"role": "user", "content": "1"},
{"role": "assistant", "content": "2"},
{"role": "user", "content": "3"},
{"role": "assistant", "content": "4"},
{"role": "user", "content": "5"},
]
groups = summarizer._find_summarization_candidates(messages, protected=set())
# Should have one group spanning all messages
assert len(groups) == 1
assert groups[0] == (0, 4)
def test_protected_splits_groups(self) -> None:
"""Protected messages split the candidates into groups."""
summarizer = ProgressiveSummarizer(min_messages_to_summarize=2)
messages = [
{"role": "user", "content": "1"},
{"role": "assistant", "content": "2"},
{"role": "user", "content": "3"}, # Protected at index 2
{"role": "assistant", "content": "4"},
{"role": "user", "content": "5"},
{"role": "assistant", "content": "6"},
]
groups = summarizer._find_summarization_candidates(messages, protected={2})
# Should have two groups: (0,1) and (3,5)
assert len(groups) == 2
assert groups[0] == (0, 1)
assert groups[1] == (3, 5)
def test_min_messages_filter(self) -> None:
"""Groups smaller than min_messages_to_summarize are filtered."""
summarizer = ProgressiveSummarizer(min_messages_to_summarize=3)
messages = [
{"role": "user", "content": "1"},
{"role": "assistant", "content": "2"},
{"role": "user", "content": "3"}, # Protected
{"role": "assistant", "content": "4"},
]
groups = summarizer._find_summarization_candidates(messages, protected={2})
# Group (0,1) has 2 messages, filtered. Group (3,3) has 1, filtered.
assert len(groups) == 0
def test_all_protected_no_candidates(self) -> None:
"""No candidates when all messages are protected."""
summarizer = ProgressiveSummarizer(min_messages_to_summarize=1)
messages = [
{"role": "user", "content": "1"},
{"role": "assistant", "content": "2"},
]
groups = summarizer._find_summarization_candidates(messages, protected={0, 1})
assert len(groups) == 0
def test_empty_messages(self) -> None:
"""Empty message list returns no groups."""
summarizer = ProgressiveSummarizer()
groups = summarizer._find_summarization_candidates([], protected=set())
assert len(groups) == 0
# =============================================================================
# Summarize Messages Tests
# =============================================================================
class TestSummarizeMessages:
"""Tests for the main summarize_messages method."""
def test_no_candidates_returns_original(
self, tokenizer: Tokenizer, simple_conversation: list[dict[str, Any]]
) -> None:
"""When no candidates, return original messages unchanged."""
summarizer = ProgressiveSummarizer(min_messages_to_summarize=100) # Too high
result = summarizer.summarize_messages(
simple_conversation, tokenizer, protected_indices=set()
)
assert len(result.messages) == len(simple_conversation)
assert result.tokens_saved == 0
assert len(result.summaries_created) == 0
def test_all_protected_no_changes(
self, tokenizer: Tokenizer, simple_conversation: list[dict[str, Any]]
) -> None:
"""All protected messages means no summarization."""
summarizer = ProgressiveSummarizer(min_messages_to_summarize=2)
all_protected = set(range(len(simple_conversation)))
result = summarizer.summarize_messages(
simple_conversation, tokenizer, protected_indices=all_protected
)
assert len(result.messages) == len(simple_conversation)
assert result.tokens_saved == 0
def test_summarization_reduces_messages(
self, tokenizer: Tokenizer, long_conversation: list[dict[str, Any]]
) -> None:
"""Summarization reduces message count."""
summarizer = ProgressiveSummarizer(
min_messages_to_summarize=3,
store_for_retrieval=False, # Skip CCR for test
)
# Protect first and last few messages
protected = {0, 1, len(long_conversation) - 1, len(long_conversation) - 2}
result = summarizer.summarize_messages(
long_conversation, tokenizer, protected_indices=protected
)
# Should have fewer messages
assert len(result.messages) < len(long_conversation)
# Should save tokens
assert result.tokens_saved > 0
# Should create summaries
assert len(result.summaries_created) > 0
def test_summarization_result_structure(
self, tokenizer: Tokenizer, long_conversation: list[dict[str, Any]]
) -> None:
"""Verify SummarizationResult has correct structure."""
summarizer = ProgressiveSummarizer(
min_messages_to_summarize=3,
store_for_retrieval=False,
)
result = summarizer.summarize_messages(long_conversation, tokenizer, protected_indices={0})
assert isinstance(result, SummarizationResult)
assert isinstance(result.messages, list)
assert isinstance(result.summaries_created, list)
assert isinstance(result.tokens_before, int)
assert isinstance(result.tokens_after, int)
assert isinstance(result.transforms_applied, list)
assert result.tokens_before >= result.tokens_after
def test_custom_summarizer_called(
self, tokenizer: Tokenizer, long_conversation: list[dict[str, Any]]
) -> None:
"""Custom summarizer function is called."""
calls: list[int] = []
def tracking_summarizer(messages: list[dict], context: str = "") -> str:
calls.append(len(messages))
return f"CUSTOM SUMMARY of {len(messages)} messages"
summarizer = ProgressiveSummarizer(
summarize_fn=tracking_summarizer,
min_messages_to_summarize=3,
store_for_retrieval=False,
)
result = summarizer.summarize_messages(long_conversation, tokenizer, protected_indices={0})
# Custom summarizer should have been called
assert len(calls) > 0
# Summary should appear in messages
found_custom = any("CUSTOM SUMMARY" in msg.get("content", "") for msg in result.messages)
assert found_custom
def test_context_passed_to_summarizer(
self, tokenizer: Tokenizer, long_conversation: list[dict[str, Any]]
) -> None:
"""Context messages are passed to summarizer."""
received_context: list[str] = []
def context_tracking_summarizer(messages: list[dict], context: str = "") -> str:
received_context.append(context)
return "Summary"
summarizer = ProgressiveSummarizer(
summarize_fn=context_tracking_summarizer,
min_messages_to_summarize=3,
store_for_retrieval=False,
)
context_msgs = [{"role": "user", "content": "Recent important question"}]
summarizer.summarize_messages(
long_conversation,
tokenizer,
protected_indices={0},
context_messages=context_msgs,
)
# Context should have been passed
assert len(received_context) > 0
# Should contain the recent message content
assert any("Recent important question" in ctx for ctx in received_context)
def test_target_tokens_stops_early(
self, tokenizer: Tokenizer, long_conversation: list[dict[str, Any]]
) -> None:
"""Summarization stops when target tokens reached."""
summarizer = ProgressiveSummarizer(
min_messages_to_summarize=3,
store_for_retrieval=False,
)
# Get original token count
original_tokens = tokenizer.count_messages(long_conversation)
# Set target very close to original (minimal summarization needed)
target = int(original_tokens * 0.95) # Only need 5% reduction
result = summarizer.summarize_messages(
long_conversation,
tokenizer,
protected_indices={0},
target_tokens=target,
)
# Should stop once target reached
assert result.tokens_after <= target or result.tokens_after < original_tokens
def test_small_groups_skipped(
self,
tokenizer: Tokenizer,
) -> None:
"""Groups with < 100 tokens are skipped."""
# Very short messages
messages = [
{"role": "user", "content": "Hi"},
{"role": "assistant", "content": "Hi"},
{"role": "user", "content": "Bye"},
{"role": "assistant", "content": "Bye"},
]
summarizer = ProgressiveSummarizer(
min_messages_to_summarize=2,
store_for_retrieval=False,
)
result = summarizer.summarize_messages(messages, tokenizer, protected_indices=set())
# Small groups should be skipped
assert len(result.summaries_created) == 0
def test_summary_larger_than_original_skipped(
self, tokenizer: Tokenizer, long_conversation: list[dict[str, Any]]
) -> None:
"""Summaries larger than original are skipped."""
def verbose_summarizer(messages: list[dict], context: str = "") -> str:
# Return a very verbose summary
return "VERY LONG SUMMARY " * 1000
summarizer = ProgressiveSummarizer(
summarize_fn=verbose_summarizer,
min_messages_to_summarize=3,
store_for_retrieval=False,
)
result = summarizer.summarize_messages(long_conversation, tokenizer, protected_indices={0})
# Summaries larger than original should be skipped
# (or if any were created, they saved tokens)
for summary in result.summaries_created:
assert summary.tokens_saved >= 0
def test_summarizer_exception_handled(
self, tokenizer: Tokenizer, long_conversation: list[dict[str, Any]]
) -> None:
"""Exceptions from summarizer are handled gracefully."""
def failing_summarizer(messages: list[dict], context: str = "") -> str:
raise ValueError("Summarization failed!")
summarizer = ProgressiveSummarizer(
summarize_fn=failing_summarizer,
min_messages_to_summarize=3,
store_for_retrieval=False,
)
# Should not raise, should return original
result = summarizer.summarize_messages(long_conversation, tokenizer, protected_indices={0})
# No summaries created due to failures
assert len(result.summaries_created) == 0
# =============================================================================
# Integration Tests
# =============================================================================
class TestProgressiveSummarizerIntegration:
"""Integration tests for end-to-end summarization."""
def test_full_workflow_with_extractive(
self, tokenizer: Tokenizer, long_conversation: list[dict[str, Any]]
) -> None:
"""Test full workflow with default extractive summarizer."""
summarizer = ProgressiveSummarizer(
min_messages_to_summarize=4,
store_for_retrieval=False,
)
original_count = len(long_conversation)
result = summarizer.summarize_messages(
long_conversation,
tokenizer,
protected_indices={0}, # Only protect system message
)
# Verify reduction
assert len(result.messages) < original_count
assert result.tokens_after < result.tokens_before
# Verify transforms tracked
assert len(result.transforms_applied) > 0
# Verify summaries created
assert len(result.summaries_created) > 0
for summary in result.summaries_created:
assert summary.start_index >= 0
assert summary.end_index >= summary.start_index
assert summary.compression_ratio < 1.0 # Actually compressed
def test_preserves_protected_messages(
self, tokenizer: Tokenizer, long_conversation: list[dict[str, Any]]
) -> None:
"""Protected messages are preserved exactly."""
summarizer = ProgressiveSummarizer(
min_messages_to_summarize=3,
store_for_retrieval=False,
)
# Protect first 3 and last 3 messages
protected = {
0,
1,
2,
len(long_conversation) - 3,
len(long_conversation) - 2,
len(long_conversation) - 1,
}
# Store original protected content
original_protected = {i: long_conversation[i]["content"] for i in protected}
result = summarizer.summarize_messages(
long_conversation,
tokenizer,
protected_indices=protected,
)
# Find protected messages in result
# First 3 should still be at beginning
assert result.messages[0]["content"] == original_protected[0]
assert result.messages[1]["content"] == original_protected[1]
assert result.messages[2]["content"] == original_protected[2]
# Last 3 should still be at end (positions shifted)
assert result.messages[-1]["content"] == original_protected[len(long_conversation) - 1]
assert result.messages[-2]["content"] == original_protected[len(long_conversation) - 2]
assert result.messages[-3]["content"] == original_protected[len(long_conversation) - 3]
def test_tool_messages_handled(
self, tokenizer: Tokenizer, conversation_with_tools: list[dict[str, Any]]
) -> None:
"""Tool messages are handled in summarization."""
# Create longer tool-heavy conversation
long_tool_conv = conversation_with_tools.copy()
for i in range(10):
long_tool_conv.extend(
[
{"role": "user", "content": f"Search again {i}"},
{
"role": "assistant",
"content": f"Searching {i}...",
"tool_calls": [
{
"id": f"call_{i}",
"type": "function",
"function": {"name": "search", "arguments": "{}"},
}
],
},
{
"role": "tool",
"tool_call_id": f"call_{i}",
"content": f'{{"data": "result {i}"}}',
},
{"role": "assistant", "content": f"Found result {i}"},
]
)
summarizer = ProgressiveSummarizer(
min_messages_to_summarize=3,
store_for_retrieval=False,
)
result = summarizer.summarize_messages(
long_tool_conv,
tokenizer,
protected_indices={0},
)
# Should reduce messages
assert len(result.messages) < len(long_tool_conv)
# Tool names should be tracked in summaries
all_tool_names = []
for summary in result.summaries_created:
all_tool_names.extend(summary.tool_names)
# Some tool calls should be tracked (may be empty if extractive)
def test_does_not_mutate_original(
self, tokenizer: Tokenizer, long_conversation: list[dict[str, Any]]
) -> None:
"""Original messages are not mutated."""
import copy
original_copy = copy.deepcopy(long_conversation)
summarizer = ProgressiveSummarizer(
min_messages_to_summarize=3,
store_for_retrieval=False,
)
summarizer.summarize_messages(
long_conversation,
tokenizer,
protected_indices={0},
)
# Original should be unchanged
assert long_conversation == original_copy
# =============================================================================
# SummarizationResult Tests
# =============================================================================
class TestSummarizationResult:
"""Tests for SummarizationResult dataclass."""
def test_tokens_saved_property(self) -> None:
"""Test tokens_saved property."""
result = SummarizationResult(
messages=[],
summaries_created=[],
tokens_before=1000,
tokens_after=300,
transforms_applied=[],
)
assert result.tokens_saved == 700
def test_tokens_saved_no_negative(self) -> None:
"""Test tokens_saved doesn't go negative."""
result = SummarizationResult(
messages=[],
summaries_created=[],
tokens_before=100,
tokens_after=150,
transforms_applied=[],
)
assert result.tokens_saved == 0