04: Next-Gen RAG
Semantic Context Compression and Knowledge Fabric
Pillar: #4 of 12 - Core Priority
Timeline: 6 Weeks
Dependencies: None
Expected ROI: 30-40% token reduction, 80% cache hits, reasoning-aware retrieval
Executive Summary
Move from document retrieval to reasoning-aware knowledge fabric. Current context management uses naive truncation that loses important early context while keeping recent noise. This pillar implements semantic compression that preserves decisions and constraints while summarizing redundant content, combined with Anthropic's prompt caching for compound savings.
Key Insight: Keep decisions and constraints verbatim; summarize explanations; drop acknowledgments entirely.
Problem Statement
Current Truncation 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 - CRITICAL):
"Build a REST API with OAuth2, JWT, rate limiting. Must be HIPAA compliant."
Message 2 (Acknowledgment - NOISE):
"Understood. I'll start working on this."
Message 3 (Status - LOW VALUE):
"I've completed the initial setup."
Message 4 (Current - KEEP):
"Here's the implementation..."
Current truncation: Drops Message 1 (most important!)
Semantic compression: Keeps Message 1, drops 2-3
Token Waste Analysis
Result: 35% of context is low/no value; 25% of critical context is lost.
Solution Architecture
Semantic Compression Pipeline
┌─────────────────┐
│ KEEP VERBATIM │ ← Decisions, constraints
│ │
Context → Semantic → Importance ───┼─────────────────┤
Analyzer Classifier │ SUMMARIZE │ ← Explanations, verbose
│ │
├─────────────────┤
│ DROP │ ← Acknowledgments, noise
│ │
└─────────────────┘
↓
┌─────────────────┐
│ CACHE BOUNDARY │ ← Stable prefix cached
│ OPTIMIZATION │ ← Dynamic suffix fresh
└─────────────────┘
Core Components
1. Semantic Compressor
class SemanticCompressor:
"""
Compress based on semantic importance, not recency.
"""
def __init__(self):
self.classifier = load_model("importance_classifier.onnx")
self.strategies = {
"decision": "verbatim",
"constraint": "verbatim",
"requirement": "verbatim",
"architecture": "verbatim",
"code": "preserve_structure",
"explanation": "summarize",
"status": "drop_or_minimal",
"acknowledgment": "drop",
"boilerplate": "drop",
}
async def compress(self, messages: List[Message], 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.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))
# "drop": don't append
# Step 3: Group summarize if still over limit
if count_tokens(compressed) > limit:
compressed = await self._group_summarize(compressed, limit)
return compressed
2. Importance Classifier
class ImportanceClassifier:
"""
Tiny classifier trained on Maestro conversation patterns.
<1ms inference on CPU.
"""
def classify(self, message: Message) -> str:
features = self._extract_features(message)
return self.model.predict(features)
def _extract_features(self, msg: Message) -> Dict:
return {
"has_code": "```" in 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"],
"persona": msg.role,
"position": msg.index / len(self.messages),
}
3. Cache-Aware Compression
class CacheAwareCompressor(SemanticCompressor):
"""
Maximize Anthropic prompt cache hits.
"""
def compress_for_cache(self, messages: List[Message], limit: int) -> List[Message]:
# Separate stable prefix from dynamic suffix
prefix, suffix = self._split_at_cache_boundary(messages)
# Keep prefix stable for cache hits (70% of limit)
stable_prefix = self._stabilize_prefix(prefix, limit * 0.7)
# Compress suffix more aggressively (30% of limit)
compressed_suffix = super().compress(suffix, limit * 0.3)
return stable_prefix + compressed_suffix
def _stabilize_prefix(self, prefix: List[Message]) -> List[Message]:
"""Remove ephemeral content (timestamps, session IDs) for cache stability."""
return [
Message(
role=msg.role,
content=self._remove_ephemeral(msg.content),
)
for msg in prefix
]
4. Knowledge Graph Integration
class KnowledgeFabric:
"""
Beyond document retrieval: reasoning-aware knowledge.
"""
def __init__(self):
self.entity_store = EntityStore()
self.relation_store = RelationStore()
self.decision_log = DecisionLog()
async def retrieve(self, query: str, context: Context) -> RetrievalResult:
# Multi-hop retrieval with entity tracking
entities = await self._extract_entities(query)
relations = await self._find_relations(entities)
# Include decision provenance
relevant_decisions = await self.decision_log.find_relevant(
query, context.workflow_id
)
return RetrievalResult(
documents=await self._semantic_search(query),
entities=entities,
relations=relations,
decisions=relevant_decisions,
reasoning_chain=self._build_chain(entities, relations),
)
async def record_decision(self, decision: Decision, context: Context):
"""Store decisions with full provenance."""
await self.decision_log.add(
decision=decision,
context=context,
entities=await self._extract_entities(decision.description),
timestamp=now(),
)
Integration Points
Files to Modify
Context Manager Modification
# BEFORE (context_manager.py)
def prepare_context(self, context: Context) -> PreparedContext:
if self.count_tokens(context) > self.limit:
return self.truncate_oldest_first(context)
return context
# AFTER
async def prepare_context(self, context: Context) -> PreparedContext:
if self.count_tokens(context) > self.limit:
return await self.semantic_compressor.compress_for_cache(
context.messages,
self.limit
)
return context
Claude Adapter Cache Integration
# Add to claude_adapter.py
async def invoke(self, context: Context) -> Response:
# Prepare with cache awareness
prepared = await self.context_manager.prepare_context(context)
# Mark cache boundaries
messages = []
for i, msg in enumerate(prepared.messages):
if i < len(prepared.messages) * 0.7: # Prefix
messages.append({
"role": msg.role,
"content": msg.content,
"cache_control": {"type": "ephemeral"}
})
else:
messages.append({"role": msg.role, "content": msg.content})
return await self.client.messages.create(
model=self.model,
messages=messages,
)
Implementation Plan
Weeks 1-2: Importance Classifier
- [ ] Extract training data from Maestro traces
- [ ] Label messages by importance category
- [ ] Train classifier (target: 90% accuracy)
- [ ] Export to ONNX for fast inference
Weeks 3-4: Semantic Compressor
- [ ] Implement SemanticCompressor class
- [ ] Add summarization using Haiku
- [ ] Implement group summarization
- [ ] Add code structure preservation
Weeks 5-6: Cache Integration & Knowledge Fabric
- [ ] Implement cache-aware compression
- [ ] Integrate Anthropic prompt caching
- [ ] Build decision log
- [ ] A/B test against truncation
- [ ] Deploy with monitoring
Expected Results
Token Reduction
Quality Preservation
Cost Savings (Combined with Caching)
Risks & Mitigations
Success Criteria
- #05 Persona Routing: FLUX + Semantic = compound savings
- #09 Knowledge Graph: Decision log feeds KG
- #03 Agent Orchestration: Knowledge propagation uses fabric
Status: Ready for Implementation
Owner: Context Optimization Team
