Semantic Context Compression: Intelligence Over Truncation
Priority: A4 (High Impact)
Timeline: 6 Weeks
Dependencies: None
The Problem with Current Truncation
Current Approach
# File: adapters/context_manager.py
def truncate_oldest_first(messages, token_limit):
"""
Drop oldest messages until under limit.
Problems:
- Loses important early context (initial requirements)
- Keeps recent noise (status updates, confirmations)
- No semantic understanding
- Binary decision: keep or drop entire message
"""
while count_tokens(messages) > token_limit:
messages.pop(0) # Drop oldest
return messages
Message 1 (Requirement - IMPORTANT):
"Build a REST API for user authentication with OAuth2, JWT tokens,
rate limiting, and audit logging. Must be HIPAA compliant."
Message 2 (Acknowledgment - LOW VALUE):
"Understood. I'll start working on the authentication API."
Message 3 (Status - LOW VALUE):
"I've completed the initial setup."
Message 4 (Status - LOW VALUE):
"Now working on the JWT implementation."
Message 5 (Current - KEEP):
"Here's the implementation so far..."
Current truncation: Drops Message 1 (most important!)
Semantic compression: Keeps Message 1, summarizes 2-4
Solution: Semantic-Aware Compression
Architecture
┌─────────────────┐
│ Keep Verbatim │ ← Decisions, constraints
│ │
Context → Semantic → Importance ───┼─────────────────┤
Analyzer Scorer │ Summarize │ ← Redundant, verbose
│ │
├─────────────────┤
│ Drop │ ← Noise, confirmations
│ │
└─────────────────┘
Implementation
class SemanticCompressor:
"""
Compress context based on semantic importance, not recency.
"""
def __init__(self):
# Classifier trained on Maestro conversation patterns
self.importance_classifier = load_model("importance_classifier.onnx")
# Categories and their retention strategies
self.retention_strategies = {
"decision": "verbatim", # Keep exactly as-is
"constraint": "verbatim", # Keep exactly as-is
"requirement": "verbatim", # Keep exactly as-is
"architecture": "verbatim", # Keep exactly as-is
"code": "preserve_structure", # Keep code, compress comments
"explanation": "summarize", # Condense to key points
"status": "drop_or_minimal", # "Phase X complete"
"acknowledgment": "drop", # Remove entirely
"boilerplate": "drop", # Remove greetings, etc.
}
async def compress(self, messages: List[Message], token_limit: int) -> List[Message]:
# Step 1: Classify each message
classified = await self._classify_messages(messages)
# Step 2: Apply retention strategies
compressed = []
for msg, category in classified:
strategy = self.retention_strategies[category]
if strategy == "verbatim":
compressed.append(msg)
elif strategy == "preserve_structure":
compressed.append(self._compress_code(msg))
elif strategy == "summarize":
compressed.append(await self._summarize(msg))
elif strategy == "drop_or_minimal":
if self._is_phase_boundary(msg):
compressed.append(self._minimal_summary(msg))
# else: drop entirely
# "drop": don't append anything
# Step 3: If still over limit, summarize groups
if count_tokens(compressed) > token_limit:
compressed = await self._group_summarize(compressed, token_limit)
return compressed
async def _classify_messages(self, messages: List[Message]) -> List[Tuple[Message, str]]:
"""
Classify each message by semantic category.
"""
results = []
for msg in messages:
features = self._extract_features(msg)
category = self.importance_classifier.predict(features)
results.append((msg, category))
return results
def _extract_features(self, msg: Message) -> Dict:
"""
Extract features for classification.
"""
return {
"has_code": bool(re.search(r'```', msg.content)),
"has_requirements": any(kw in msg.content.lower()
for kw in ["must", "should", "requirement", "constraint"]),
"has_decision": any(kw in msg.content.lower()
for kw in ["decided", "choosing", "will use", "approach"]),
"is_short": len(msg.content) < 100,
"is_acknowledgment": msg.content.strip().lower() in
["ok", "understood", "got it", "sure", "will do"],
"persona": msg.role,
"position": msg.index / len(messages), # Relative position
}
async def _summarize(self, msg: Message) -> Message:
"""
Summarize verbose explanations to key points.
Uses small model (Haiku) for speed.
"""
summary = await self.summarizer.summarize(
msg.content,
max_tokens=100,
style="bullet_points"
)
return Message(
role=msg.role,
content=f"[Summary] {summary}",
metadata={**msg.metadata, "compressed": True}
)
async def _group_summarize(self, messages: List[Message], limit: int) -> List[Message]:
"""
Group similar messages and summarize together.
"""
# Group consecutive messages of same type
groups = self._group_consecutive(messages)
compressed = []
for group in groups:
if len(group) == 1:
compressed.append(group[0])
else:
# Summarize the group
combined = "\n".join(m.content for m in group)
summary = await self._summarize(Message(
role=group[0].role,
content=combined
))
compressed.append(summary)
if count_tokens(compressed) >= limit * 0.9:
break
return compressed
Training the Importance Classifier
Data Collection
# Generate training data from Maestro traces
def generate_training_data(traces):
training_data = []
for trace in traces:
for msg_idx, message in enumerate(trace.messages):
# Label based on what humans/reviewers considered important
label = determine_importance(message, trace)
training_data.append({
"content": message.content[:500], # Truncate for features
"has_code": "```" in message.content,
"has_requirements": contains_requirements(message),
"has_decision": contains_decision(message),
"is_short": len(message.content) < 100,
"position": msg_idx / len(trace.messages),
"persona": message.role,
# Label
"category": label, # decision, constraint, status, etc.
})
return training_data
def determine_importance(message, trace):
"""
Heuristic labeling based on workflow outcomes.
"""
# Messages referenced in final output = important
if message_referenced_in_output(message, trace.output):
return "requirement" if contains_requirements(message) else "decision"
# Short acknowledgments
if len(message.content) < 50 and is_acknowledgment(message):
return "acknowledgment"
# Status updates without new information
if is_status_update(message) and not contains_new_info(message):
return "status"
# Default to explanation (summarizable)
return "explanation"
Model Architecture
class ImportanceClassifier(nn.Module):
"""
Tiny classifier for message importance.
Runs in <1ms on CPU.
"""
def __init__(self, num_categories=8):
super().__init__()
# Text encoder (small transformer or TF-IDF)
self.text_encoder = nn.Sequential(
nn.Embedding(10000, 64), # Vocab size, embed dim
nn.TransformerEncoderLayer(d_model=64, nhead=4),
nn.AdaptiveAvgPool1d(1),
)
# Feature encoder
self.feature_encoder = nn.Linear(10, 32)
# Classifier
self.classifier = nn.Sequential(
nn.Linear(64 + 32, 64),
nn.ReLU(),
nn.Dropout(0.1),
nn.Linear(64, num_categories),
)
def forward(self, text_ids, features):
text_emb = self.text_encoder(text_ids)
feat_emb = self.feature_encoder(features)
combined = torch.cat([text_emb, feat_emb], dim=-1)
return self.classifier(combined)
Integration with Anthropic Prompt Caching
Maximize Cache Hits
class CacheAwareCompressor(SemanticCompressor):
"""
Compress while maximizing prompt cache effectiveness.
"""
def compress_for_cache(self, messages: List[Message], token_limit: int) -> List[Message]:
# Separate into cacheable prefix and dynamic suffix
prefix, suffix = self._split_at_cache_boundary(messages)
# Compress suffix more aggressively (changes each call)
compressed_suffix = super().compress(suffix, token_limit * 0.3)
# Keep prefix stable for cache hits (changes rarely)
stable_prefix = self._stabilize_prefix(prefix, token_limit * 0.7)
return stable_prefix + compressed_suffix
def _split_at_cache_boundary(self, messages: List[Message]) -> Tuple[List, List]:
"""
Find natural boundary between stable and dynamic content.
"""
# System prompt + initial requirements = stable
# Recent conversation = dynamic
for i, msg in enumerate(messages):
if msg.metadata.get("phase_boundary"):
return messages[:i], messages[i:]
# Default: first 70% is prefix
split_point = int(len(messages) * 0.7)
return messages[:split_point], messages[split_point:]
def _stabilize_prefix(self, prefix: List[Message], token_limit: int) -> List[Message]:
"""
Ensure prefix is identical across calls for cache hits.
"""
# Remove timestamps, session IDs, etc.
stabilized = []
for msg in prefix:
clean = Message(
role=msg.role,
content=self._remove_ephemeral_content(msg.content),
)
stabilized.append(clean)
return stabilized
Cache Control Integration
class CacheOptimizedAdapter(ClaudeAdapter):
"""
Use Anthropic's prompt caching with semantic compression.
"""
async def invoke(self, context: Context) -> Response:
# Compress with cache awareness
compressed = self.compressor.compress_for_cache(
context.messages,
token_limit=context.max_tokens
)
# Mark cache boundaries
messages = []
for i, msg in enumerate(compressed):
if i < len(compressed) * 0.7: # Prefix
messages.append({
"role": msg.role,
"content": msg.content,
"cache_control": {"type": "ephemeral"} # Cache this
})
else:
messages.append({
"role": msg.role,
"content": msg.content,
})
response = await self.client.messages.create(
model=self.model,
messages=messages,
# ... other params
)
return response
Expected Results
Token Reduction
Quality Preservation
Cost Savings
6-Week Implementation Plan
Weeks 1-2: Classifier Training
- [ ] Extract training data from Maestro traces
- [ ] Label messages by importance category
- [ ] Train and validate importance classifier
- [ ] Export to ONNX for fast inference
Weeks 3-4: Compression Engine
- [ ] Implement SemanticCompressor
- [ ] Add summarization with Haiku
- [ ] Implement group summarization
- [ ] Add code structure preservation
Weeks 5-6: Cache Integration
- [ ] Implement cache-aware compression
- [ ] Integrate with Anthropic prompt caching
- [ ] A/B test against truncation
- [ ] Deploy and monitor
Status: Ready for Implementation
Key Insight: Keep decisions and constraints, summarize explanations, drop noise
Owner: Context Optimization Team
