
Context Optimization
- 114 installs
- 941 repo stars
- Updated August 5, 2026
- guanyang/antigravity-skills
Shrink or prioritize agent context—summaries, selective file loads, deduped tool results—when sessions hit token limits or latency grows on large codebases.
About
Context-optimization teaches strategies to trim, rank, and compress agent context—summaries, selective retrieval, and deduplicated tool results—so long coding sessions stay within token limits while preserving task-critical information.
- Prioritizes high-signal context slices
- Compresses logs and duplicate tool output
- Balances recall vs token spend
- Supports rolling summaries and checkpoints
- Reduces latency on large monorepos
Context Optimization by the numbers
- 114 all-time installs (skills.sh)
- Ranked #3,952 of 16,546 AI & Agent Building skills by installs in the Skillselion catalog
- Data as of Aug 5, 2026 (Skillselion catalog sync)
npx skills add https://github.com/guanyang/antigravity-skills --skill context-optimizationAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 114 |
|---|---|
| repo stars | ★ 941 |
| Last updated | August 5, 2026 |
| Repository | guanyang/antigravity-skills ↗ |
What it does
Shrink or prioritize agent context—summaries, selective file loads, deduped tool results—when sessions hit token limits or latency grows on large codebases.
Files
Context Optimization Techniques
Context optimization extends the effective capacity of limited context windows through strategic compression, masking, caching, and partitioning. Effective optimization can double or triple effective context capacity without requiring larger models or longer windows — but only when applied with discipline. The techniques below are ordered by impact and risk.
When to Activate
Activate this skill when:
- Context limits constrain task complexity
- Optimizing for cost reduction (fewer tokens = lower costs)
- Reducing latency for long conversations
- Implementing long-running agent systems
- Needing to handle larger documents or conversations
- Building production systems at scale
Core Concepts
Apply four primary strategies in this priority order:
1. KV-cache optimization — Reorder and stabilize prompt structure so the inference engine reuses cached Key/Value tensors. This is the cheapest optimization: zero quality risk, immediate cost and latency savings. Apply it first and unconditionally.
2. Observation masking — Replace verbose tool outputs with compact references once their purpose has been served. Tool outputs consume 80%+ of tokens in typical agent trajectories, so masking them yields the largest capacity gains. The original content remains retrievable if needed downstream.
3. Compaction — Summarize accumulated context when utilization exceeds 70%, then reinitialize with the summary. This distills the window's contents while preserving task-critical state. Compaction is lossy — apply it after masking has already removed the low-value bulk.
4. Context partitioning — Split work across sub-agents with isolated contexts when a single window cannot hold the full problem. Each sub-agent operates in a clean context focused on its subtask. Reserve this for tasks where estimated context exceeds 60% of the window limit, because coordination overhead is real.
The governing principle: context quality matters more than quantity. Every optimization preserves signal while reducing noise. Measure before optimizing, then measure the optimization's effect.
Detailed Topics
Compaction Strategies
Trigger compaction when context utilization exceeds 70%: summarize the current context, then reinitialize with the summary. This distills the window's contents in a high-fidelity manner, enabling continuation with minimal performance degradation. Prioritize compressing tool outputs first (they consume 80%+ of tokens), then old conversation turns, then retrieved documents. Never compress the system prompt — it anchors model behavior and its removal causes unpredictable degradation.
Preserve different elements by message type:
- Tool outputs: Extract key findings, metrics, error codes, and conclusions. Strip verbose raw output, stack traces (unless debugging is ongoing), and boilerplate headers.
- Conversational turns: Retain decisions, commitments, user preferences, and context shifts. Remove filler, pleasantries, and exploratory back-and-forth that led to a conclusion already captured.
- Retrieved documents: Keep claims, facts, and data points relevant to the active task. Remove supporting evidence and elaboration that served a one-time reasoning purpose.
Target 50-70% token reduction with less than 5% quality degradation. If compaction exceeds 70% reduction, audit the summary for critical information loss — over-aggressive compaction is the most common failure mode.
Observation Masking
Mask observations selectively based on recency and ongoing relevance — not uniformly. Apply these rules:
- Never mask: Observations critical to the current task, observations from the most recent turn, observations used in active reasoning chains, and error outputs when debugging is in progress.
- Mask after 3+ turns: Verbose outputs whose key points have already been extracted into the conversation flow. Replace with a compact reference:
[Obs:{ref_id} elided. Key: {summary}. Full content retrievable.] - Always mask immediately: Repeated/duplicate outputs, boilerplate headers and footers, outputs already summarized earlier in the conversation.
Masking should achieve 60-80% reduction in masked observations with less than 2% quality impact. The key is maintaining retrievability — store the full content externally and keep the reference ID in context so the agent can request the original if needed.
KV-Cache Optimization
Maximize prefix cache hits by structuring prompts so that stable content occupies the prefix and dynamic content appears at the end. KV-cache stores Key and Value tensors computed during inference; when consecutive requests share an identical prefix, the cached tensors are reused, saving both cost and latency.
Apply this ordering in every prompt: 1. System prompt (most stable — never changes within a session) 2. Tool definitions (stable across requests) 3. Frequently reused templates and few-shot examples 4. Conversation history (grows but shares prefix with prior turns) 5. Current query and dynamic content (least stable — always last)
Design prompts for cache stability: remove timestamps, session counters, and request IDs from the system prompt. Move dynamic metadata into a separate user message or tool result where it does not break the prefix. Even a single whitespace change in the prefix invalidates the entire cached block downstream of that change.
Target 70%+ cache hit rate for stable workloads. At scale, this translates to 50%+ cost reduction and 40%+ latency reduction on cached tokens.
Context Partitioning
Partition work across sub-agents when a single context cannot hold the full problem without triggering aggressive compaction. Each sub-agent operates in a clean, focused context for its subtask, then returns a structured result to a coordinator agent.
Plan partitioning when estimated task context exceeds 60% of the window limit. Decompose the task into independent subtasks, assign each to a sub-agent, and aggregate results. Validate that all partitions completed before merging, merge compatible results, and apply summarization if the aggregated output still exceeds budget.
This approach achieves separation of concerns — detailed search context stays isolated within sub-agents while the coordinator focuses on synthesis. However, coordination has real token cost: the coordinator prompt, result aggregation, and error handling all consume tokens. Only partition when the savings exceed this overhead.
Budget Management
Allocate explicit token budgets across context categories before the session begins: system prompt, tool definitions, retrieved documents, message history, tool outputs, and a reserved buffer (5-10% of total). Monitor usage against budget continuously and trigger optimization when any category exceeds its allocation or total utilization crosses 70%.
Use trigger-based optimization rather than periodic optimization. Monitor these signals:
- Token utilization above 80% — trigger compaction
- Attention degradation indicators (repetition, missed instructions) — trigger masking + compaction
- Quality score drops below baseline — audit context composition before optimizing
Practical Guidance
Optimization Decision Framework
Select the optimization technique based on what dominates the context:
| Context Composition | First Action | Second Action |
|---|---|---|
| Tool outputs dominate (>50%) | Observation masking | Compaction of remaining turns |
| Retrieved documents dominate | Summarization | Partitioning if docs are independent |
| Message history dominates | Compaction with selective preservation | Partitioning for new subtasks |
| Multiple components contribute | KV-cache optimization first, then layer masking + compaction | |
| Near-limit with active debugging | Mask resolved tool outputs only — preserve error details |
Performance Targets
Track these metrics to validate optimization effectiveness:
- Compaction: 50-70% token reduction, <5% quality degradation, <10% latency overhead from the compaction step itself
- Masking: 60-80% reduction in masked observations, <2% quality impact, near-zero latency overhead
- Cache optimization: 70%+ hit rate for stable workloads, 50%+ cost reduction, 40%+ latency reduction
- Partitioning: Net token savings after accounting for coordinator overhead; break-even typically requires 3+ subtasks
Iterate on strategies based on measured results. If an optimization technique does not measurably improve the target metric, remove it — optimization machinery itself consumes tokens and adds latency.
Examples
Example 1: Compaction Trigger
if context_tokens / context_limit > 0.8:
context = compact_context(context)Example 2: Observation Masking
if len(observation) > max_length:
ref_id = store_observation(observation)
return f"[Obs:{ref_id} elided. Key: {extract_key(observation)}]"Example 3: Cache-Friendly Ordering
# Stable content first
context = [system_prompt, tool_definitions] # Cacheable
context += [reused_templates] # Reusable
context += [unique_content] # UniqueGuidelines
1. Measure before optimizing—know your current state 2. Apply masking before compaction — remove low-value bulk first, then summarize what remains 3. Design for cache stability with consistent prompts 4. Partition before context becomes problematic 5. Monitor optimization effectiveness over time 6. Balance token savings against quality preservation 7. Test optimization at production scale 8. Implement graceful degradation for edge cases
Gotchas
1. Whitespace breaks KV-cache: Even a single whitespace or newline change in the prompt prefix invalidates the entire KV-cache block downstream of that point. Pin system prompts as immutable strings — do not interpolate timestamps, version numbers, or session IDs into them. Diff prompt templates byte-for-byte between deployments.
2. Timestamps in system prompts destroy cache hit rates: Including Current date: {today} or similar dynamic content in the system prompt forces a full cache miss on every new day (or every request, if using time-of-day). Move dynamic metadata into a user message or a separate tool result appended after the stable prefix.
3. Compaction under pressure loses critical state: When the model performing compaction is itself under context pressure (>85% utilization), its summarization quality degrades — it omits task goals, drops user constraints, and flattens nuanced state. Trigger compaction at 70-80%, not 90%+. If compaction must happen late, use a separate model call with a clean context containing only the material to summarize.
4. Masking error outputs breaks debugging loops: Over-aggressive masking hides error messages, stack traces, and failure details that the agent needs in subsequent turns to diagnose and fix issues. During active debugging (error in the last 3 turns), suspend masking for all error-related observations until the issue is resolved.
5. Partitioning overhead can exceed savings: Each sub-agent requires its own system prompt, tool definitions, and coordination messages. For tasks with fewer than 3 independent subtasks, the coordination overhead often exceeds the context savings. Estimate total tokens (coordinator + all sub-agents) before committing to partitioning.
6. Cache miss cost spikes after deployment changes: Reordering tools, rewording the system prompt, or changing few-shot examples between deployments invalidates the entire prefix cache, causing a temporary cost spike of 2-5x until the new cache warms up. Roll out prompt changes gradually and monitor cache hit rate during deployment windows.
7. Compaction creates false confidence in stale summaries: Once context is compacted, the summary looks authoritative but may reflect outdated state. If the task has evolved since compaction (new user requirements, corrected assumptions), the summary silently carries forward stale information. After compaction, re-validate the summary against the current task goal before proceeding.
Integration
This skill builds on context-fundamentals and context-degradation. It connects to:
- multi-agent-patterns - Partitioning as isolation
- latent-briefing - Selective KV retention across orchestrator–worker boundaries (compatible models)
- evaluation - Measuring optimization effectiveness
- memory-systems - Offloading context to memory
References
Internal reference:
- Optimization Techniques Reference - Read when: implementing a specific optimization technique and needing detailed code patterns, threshold tables, or integration examples beyond what the skill body provides
Related skills in this collection:
- context-fundamentals - Read when: unfamiliar with context window mechanics, token counting, or attention distribution basics
- context-degradation - Read when: diagnosing why agent performance has dropped and needing to identify which degradation pattern is occurring before selecting an optimization
- evaluation - Read when: setting up metrics and benchmarks to measure whether an optimization technique actually improved outcomes
External resources:
- Research on context window limitations - Read when: evaluating model-specific context behavior (e.g., lost-in-the-middle effects, attention decay curves)
- KV-cache optimization techniques - Read when: implementing prefix caching at the inference infrastructure level (vLLM, TGI, or cloud provider APIs)
- Production engineering guides - Read when: deploying context optimization in a production pipeline and needing operability patterns (monitoring, alerting, rollback)
---
Skill Metadata
Created: 2025-12-20 Last Updated: 2026-03-17 Author: Agent Skills for Context Engineering Contributors Version: 2.0.0
Context Optimization Reference
This document provides detailed technical reference for context optimization techniques and strategies.
Compaction Strategies
Summary-Based Compaction
Summary-based compaction replaces verbose content with concise summaries while preserving key information. The approach works by identifying sections that can be compressed, generating summaries that capture essential points, and replacing full content with summaries.
The effectiveness of compaction depends on what information is preserved. Critical decisions, user preferences, and current task state should never be compacted. Intermediate results and supporting evidence can be summarized more aggressively. Boilerplate, repeated information, and exploratory reasoning can often be removed entirely.
Token Budget Allocation
Effective context budgeting requires understanding how different context components consume tokens and allocating budget strategically:
| Component | Typical Range | Notes |
|---|---|---|
| System prompt | 500-2000 tokens | Stable across session |
| Tool definitions | 100-500 per tool | Grows with tool count |
| Retrieved documents | Variable | Often largest consumer |
| Message history | Variable | Grows with conversation |
| Tool outputs | Variable | Can dominate context |
Compaction Thresholds
Trigger compaction at appropriate thresholds to maintain performance:
- Warning threshold at 70% of effective context limit
- Compaction trigger at 80% of effective context limit
- Aggressive compaction at 90% of effective context limit
The exact thresholds depend on model behavior and task characteristics. Some models show graceful degradation while others exhibit sharp performance cliffs.
Observation Masking Patterns
Selective Masking
Not all observations should be masked equally. Consider masking observations that have served their purpose and are no longer needed for active reasoning. Keep observations that are central to the current task. Keep observations from the most recent turn. Keep observations that may be referenced again.
Masking Implementation
def selective_mask(observations: List[Dict], current_task: Dict) -> List[Dict]:
"""
Selectively mask observations based on relevance.
Returns observations with mask field indicating masked content.
"""
masked = []
for obs in observations:
relevance = calculate_relevance(obs, current_task)
if relevance < 0.3 and obs["age"] > 3:
# Low relevance and old - mask
masked.append({
**obs,
"masked": True,
"reference": store_for_reference(obs["content"]),
"summary": summarize_content(obs["content"])
})
else:
masked.append({
**obs,
"masked": False
})
return maskedKV-Cache Optimization
Prefix Stability
KV-cache hit rates depend on prefix stability. Stable prefixes enable cache reuse across requests. Dynamic prefixes invalidate cache and force recomputation.
Elements that should remain stable include system prompts, tool definitions, and frequently used templates. Elements that may vary include timestamps, session identifiers, and query-specific content.
Cache-Friendly Design
Design prompts to maximize cache hit rates:
1. Place stable content at the beginning 2. Use consistent formatting across requests 3. Avoid dynamic content in prompts when possible 4. Use placeholders for dynamic content
# Cache-unfriendly: Dynamic timestamp in prompt
system_prompt = f"""
Current time: {datetime.now().isoformat()}
You are a helpful assistant.
"""
# Cache-friendly: Stable prompt with dynamic time as variable
system_prompt = """
You are a helpful assistant.
Current time is provided separately when relevant.
"""Context Partitioning Strategies
Sub-Agent Isolation
Partition work across sub-agents to prevent any single context from growing too large. Each sub-agent operates with a clean context focused on its subtask.
Partition Planning
def plan_partitioning(task: Dict, context_limit: int) -> Dict:
"""
Plan how to partition a task based on context limits.
Returns partitioning strategy and subtask definitions.
"""
estimated_context = estimate_task_context(task)
if estimated_context <= context_limit:
return {
"strategy": "single_agent",
"subtasks": [task]
}
# Plan multi-agent approach
subtasks = decompose_task(task)
return {
"strategy": "multi_agent",
"subtasks": subtasks,
"coordination": "hierarchical"
}Optimization Decision Framework
When to Optimize
Consider context optimization when context utilization exceeds 70%, when response quality degrades as conversations extend, when costs increase due to long contexts, or when latency increases with conversation length.
What Optimization to Apply
Choose optimization strategies based on context composition:
If tool outputs dominate context, apply observation masking. If retrieved documents dominate context, apply summarization or partitioning. If message history dominates context, apply compaction with summarization. If multiple components contribute, combine strategies.
Evaluation of Optimization
After applying optimization, evaluate effectiveness:
- Measure token reduction achieved
- Measure quality preservation (output quality should not degrade)
- Measure latency improvement
- Measure cost reduction
Iterate on optimization strategies based on evaluation results.
Common Pitfalls
Over-Aggressive Compaction
Compacting too aggressively can remove critical information. Always preserve task goals, user preferences, and recent conversation context. Test compaction at increasing aggressiveness levels to find the optimal balance.
Masking Critical Observations
Masking observations that are still needed can cause errors. Track observation usage and only mask content that is no longer referenced. Consider keeping references to masked content that could be retrieved if needed.
Ignoring Attention Distribution
The lost-in-middle phenomenon means that information placement matters. Place critical information at attention-favored positions (beginning and end of context). Use explicit markers to highlight important content.
Premature Optimization
Not all contexts require optimization. Adding optimization machinery has overhead. Optimize only when context limits actually constrain agent performance.
Monitoring and Alerting
Key Metrics
Track these metrics to understand optimization needs:
- Context token count over time
- Cache hit rates for repeated patterns
- Response quality metrics by context size
- Cost per conversation by context length
- Latency by context size
Alert Thresholds
Set alerts for:
- Context utilization above 80%
- Cache hit rate below 50%
- Quality score drop of more than 10%
- Cost increase above baseline
Integration Patterns
Integration with Agent Framework
Integrate optimization into agent workflow:
class OptimizingAgent:
def __init__(self, context_limit: int = 80000):
self.context_limit = context_limit
self.optimizer = ContextOptimizer()
def process(self, user_input: str, context: Dict) -> Dict:
# Check if optimization needed
if self.optimizer.should_compact(context):
context = self.optimizer.compact(context)
# Process with optimized context
response = self._call_model(user_input, context)
# Track metrics
self.optimizer.record_metrics(context, response)
return responseIntegration with Memory Systems
Connect optimization with memory systems:
class MemoryAwareOptimizer:
def __init__(self, memory_system, context_limit: int):
self.memory = memory_system
self.limit = context_limit
def optimize_context(self, current_context: Dict, task: str) -> Dict:
# Check if information is in memory
relevant_memories = self.memory.retrieve(task)
# Move information to memory if not needed in context
for mem in relevant_memories:
if mem["importance"] < threshold:
current_context = remove_from_context(current_context, mem)
# Keep reference that memory can be retrieved
return current_contextPerformance Benchmarks
Compaction Performance
Compaction should reduce token count while preserving quality. Target:
- 50-70% token reduction for aggressive compaction
- Less than 5% quality degradation from compaction
- Less than 10% latency increase from compaction overhead
Masking Performance
Observation masking should reduce token count significantly:
- 60-80% reduction in masked observations
- Less than 2% quality impact from masking
- Near-zero latency overhead
Cache Performance
KV-cache optimization should improve cost and latency:
- 70%+ cache hit rate for stable workloads
- 50%+ cost reduction from cache hits
- 40%+ latency reduction from cache hits
"""
Context Optimization Utilities — compaction, masking, budgeting, and cache optimization.
Public API
----------
Functions:
estimate_token_count(text) -> int
estimate_message_tokens(messages) -> int
categorize_messages(messages) -> dict
summarize_content(content, category, max_length) -> str
design_stable_prompt(template, dynamic_values) -> str
calculate_cache_metrics(requests, cache) -> dict
Classes:
ObservationStore — Store and mask verbose tool outputs with retrievable references.
ContextBudget — Token budget allocation and optimization trigger detection.
PRODUCTION NOTES:
- Token estimation uses simplified heuristics (~4 chars/token for English).
Production systems should use model-specific tokenizers:
- OpenAI: tiktoken library
- Anthropic: anthropic tokenizer
- Local models: HuggingFace tokenizers
- Summarization functions use simple heuristics for demonstration.
Production systems should use:
- LLM-based summarization for high-quality compression
- Domain-specific summarization models
- Schema-based summarization for structured outputs
- Cache metrics are illustrative. Production systems should integrate
with actual inference infrastructure metrics.
"""
from typing import List, Dict, Optional, Tuple
import hashlib
import re
import time
__all__ = [
"estimate_token_count",
"estimate_message_tokens",
"categorize_messages",
"summarize_content",
"summarize_tool_output",
"summarize_conversation",
"summarize_document",
"summarize_general",
"ObservationStore",
"ContextBudget",
"design_stable_prompt",
"calculate_cache_metrics",
"generate_cache_recommendations",
]
# ---------------------------------------------------------------------------
# Token estimation
# ---------------------------------------------------------------------------
def estimate_token_count(text: str) -> int:
"""
Estimate token count for text.
Use when: a quick token budget check is needed and a model-specific
tokenizer is unavailable or too slow for the hot path.
Uses approximation: ~4 characters per token for English.
WARNING: This is a rough estimate. Actual tokenization varies by:
- Model (GPT-5.2, Claude 4.5, Gemini 3 have different tokenizers)
- Content type (code typically has higher token density)
- Language (non-English may have 2-3x higher token/char ratio)
Production usage::
import tiktoken
enc = tiktoken.encoding_for_model("gpt-4")
token_count = len(enc.encode(text))
"""
return len(text) // 4
def estimate_message_tokens(messages: List[Dict[str, str]]) -> int:
"""
Estimate token count for a message list.
Use when: checking whether the current conversation is approaching
the context budget threshold before deciding to compact or mask.
"""
total = 0
for msg in messages:
content = msg.get("content", "")
total += estimate_token_count(content)
# Add overhead for role/formatting
total += 10
return total
# ---------------------------------------------------------------------------
# Compaction functions
# ---------------------------------------------------------------------------
def categorize_messages(messages: List[Dict]) -> Dict[str, List[Dict]]:
"""
Categorize messages for selective compaction.
Use when: preparing to compact context and needing to apply different
summarization strategies per category (tool outputs first, then old
conversation turns, then retrieved documents — never the system prompt).
Returns a dict mapping category name to list of messages.
"""
categories: Dict[str, List[Dict]] = {
"system_prompt": [],
"tool_definition": [],
"tool_output": [],
"conversation": [],
"retrieved_document": [],
"other": [],
}
for msg in messages:
role = msg.get("role", "user")
content = msg.get("content", "")
if role == "system":
categories["system_prompt"].append({**msg, "category": "system_prompt"})
elif "tool_use" in msg.get("type", ""):
categories["tool_output"].append({**msg, "category": "tool_output"})
elif role == "user":
categories["conversation"].append({**msg, "category": "conversation"})
elif "retrieved" in msg.get("tags", []):
categories["retrieved_document"].append({**msg, "category": "retrieved_document"})
else:
categories["other"].append({**msg, "category": "other"})
return categories
def summarize_content(content: str, category: str, max_length: int = 500) -> str:
"""
Summarize content for compaction, dispatching by category.
Use when: compacting context and needing category-aware summarization
(tool outputs get metric extraction, conversations get decision
extraction, documents get lead-paragraph extraction).
"""
if category == "tool_output":
return summarize_tool_output(content, max_length)
elif category == "conversation":
return summarize_conversation(content, max_length)
elif category == "retrieved_document":
return summarize_document(content, max_length)
else:
return summarize_general(content, max_length)
def summarize_tool_output(content: str, max_length: int = 500) -> str:
"""
Summarize tool output by extracting metrics and key findings.
Use when: a tool output has served its immediate purpose and needs
to be compacted while preserving actionable data points.
"""
# Look for metrics (numbers with context)
metrics = re.findall(r'(\w+):\s*([\d.,]+)', content)
# Look for key findings (lines with important keywords)
keywords = ["result", "found", "total", "success", "error", "value"]
findings = []
for line in content.split('\n'):
if any(kw in line.lower() for kw in keywords):
findings.append(line.strip())
summary_parts = []
if metrics:
summary_parts.append(f"Metrics: {', '.join([f'{k}={v}' for k, v in metrics])}")
if findings:
summary_parts.append("Key findings: " + "; ".join(findings[:3]))
result = " | ".join(summary_parts) if summary_parts else "[Tool output summarized]"
return result[:max_length]
def summarize_conversation(content: str, max_length: int = 500) -> str:
"""
Summarize conversational content by extracting decisions and questions.
Use when: older conversation turns need compaction and the key
decisions/commitments must survive while filler is removed.
"""
decisions = re.findall(r'(?i)(?:decided|decision|chose|chosen)[:\s]+([^.]+)', content)
questions = re.findall(r'(?:\?|question)[:\s]+([^.]+)', content)
summary_parts = []
if decisions:
decision_texts = [d.strip() for d in decisions[:5]]
summary_parts.append(f"Decisions: {'; '.join(decision_texts)}")
if questions:
question_texts = [q.strip() for q in questions[:3]]
summary_parts.append(f"Open questions: {'; '.join(question_texts)}")
if not summary_parts:
# Fallback: extract the first few substantive sentences
sentences = [s.strip() for s in content.split('.') if len(s.strip()) > 20]
if sentences:
summary_parts.append('. '.join(sentences[:3]) + '.')
result = " | ".join(summary_parts) if summary_parts else "[Conversation summarized]"
return result[:max_length]
def summarize_document(content: str, max_length: int = 500) -> str:
"""
Summarize document content using lead-paragraph extraction.
Use when: a retrieved document has been consumed for reasoning and
only a brief reference needs to remain in context.
"""
paragraphs = content.split('\n\n')
if paragraphs:
first_para = paragraphs[0].strip()
sentences = first_para.split('. ')
if len(sentences) > 2:
first_para = '. '.join(sentences[:2]) + '.'
return first_para[:max_length]
return "[Document summarized]"
def summarize_general(content: str, max_length: int = 500) -> str:
"""
General-purpose summarization via truncation.
Use when: content does not fit a specific category and a simple
truncation with ellipsis is acceptable.
"""
return content[:max_length] + "..." if len(content) > max_length else content
# ---------------------------------------------------------------------------
# Observation masking
# ---------------------------------------------------------------------------
class ObservationStore:
"""
Store and mask verbose tool outputs with retrievable references.
Use when: tool outputs dominate context (>50% of tokens) and older
observations have already served their reasoning purpose. Stores the
full content externally and replaces it with a compact reference
containing a key-point summary.
Example::
store = ObservationStore(max_size=500)
masked, ref_id = store.mask(long_tool_output, max_length=200)
# masked: "[Obs:a1b2c3d4 elided. Key: ... Full content retrievable.]"
# Later retrieval:
original = store.retrieve(ref_id)
"""
def __init__(self, max_size: int = 1000) -> None:
self.observations: Dict[str, Dict] = {}
self.order: List[str] = []
self.max_size = max_size
def store(self, content: str, metadata: Optional[Dict] = None) -> str:
"""Store observation and return reference ID."""
ref_id = self._generate_ref_id(content)
self.observations[ref_id] = {
"content": content,
"metadata": metadata or {},
"stored_at": time.time(),
"last_accessed": time.time(),
}
self.order.append(ref_id)
# Evict oldest if over limit
if len(self.order) > self.max_size:
oldest = self.order.pop(0)
del self.observations[oldest]
return ref_id
def retrieve(self, ref_id: str) -> Optional[str]:
"""Retrieve observation by reference ID."""
if ref_id in self.observations:
self.observations[ref_id]["last_accessed"] = time.time()
return self.observations[ref_id]["content"]
return None
def mask(self, content: str, max_length: int = 200) -> Tuple[str, Optional[str]]:
"""
Mask observation if longer than max_length.
Use when: deciding per-observation whether to keep inline or
replace with a compact reference. Returns (masked_content, ref_id)
where ref_id is None if the content was short enough to keep.
"""
if len(content) <= max_length:
return content, None
ref_id = self.store(content)
key_point = self._extract_key_point(content)
masked = f"[Obs:{ref_id} elided. Key: {key_point}. Full content retrievable.]"
return masked, ref_id
def _generate_ref_id(self, content: str) -> str:
"""Generate unique reference ID."""
hash_input = f"{content[:100]}{time.time()}"
return hashlib.md5(hash_input.encode()).hexdigest()[:8]
def _extract_key_point(self, content: str) -> str:
"""Extract key point from observation."""
lines = [line for line in content.split('\n') if len(line) > 20]
if lines:
return lines[0][:50] + "..."
sentences = content.split('. ')
if sentences:
return sentences[0][:50] + "..."
return content[:50] + "..."
# ---------------------------------------------------------------------------
# Context budget management
# ---------------------------------------------------------------------------
class ContextBudget:
"""
Token budget allocation and optimization trigger detection.
Use when: building an agent loop that needs to monitor context usage
across categories and trigger compaction/masking at the right thresholds
rather than waiting until the window overflows.
Example::
budget = ContextBudget(total_limit=128_000)
budget.allocate("system_prompt", 1500)
budget.allocate("tool_definitions", 3000)
# ... after each agent turn:
should_act, reasons = budget.should_optimize(current_usage)
if should_act:
# apply masking or compaction based on reasons
pass
"""
def __init__(self, total_limit: int) -> None:
self.total_limit = total_limit
self.allocated: Dict[str, int] = {
"system_prompt": 0,
"tool_definitions": 0,
"retrieved_docs": 0,
"message_history": 0,
"tool_outputs": 0,
"other": 0,
}
self.reserved = 5000 # Reserved buffer
self.reservation_limit = total_limit - self.reserved
def allocate(self, category: str, amount: int) -> bool:
"""
Allocate budget to category. Returns True on success, False if
the allocation would exceed the reservation limit.
"""
if category not in self.allocated:
category = "other"
current = sum(self.allocated.values())
proposed = current + amount
if proposed > self.reservation_limit:
return False
self.allocated[category] += amount
return True
def remaining(self) -> int:
"""Get remaining unallocated budget."""
current = sum(self.allocated.values())
return self.reservation_limit - current
def get_usage(self) -> Dict[str, object]:
"""
Get current usage breakdown.
Use when: logging or displaying context budget state for
monitoring dashboards or debug output.
"""
total = sum(self.allocated.values())
return {
"total_used": total,
"total_limit": self.total_limit,
"remaining": self.remaining(),
"by_category": dict(self.allocated),
"utilization_ratio": total / self.total_limit,
}
def should_optimize(
self, current_usage: int, metrics: Optional[Dict[str, float]] = None
) -> Tuple[bool, List[Tuple[str, object]]]:
"""
Determine if optimization should trigger.
Use when: called at the end of each agent loop iteration to
decide whether to apply compaction, masking, or both before
the next model call.
Returns (should_optimize, list_of_reasons).
"""
reasons: List[Tuple[str, object]] = []
# Check utilization
utilization = current_usage / self.total_limit
if utilization > 0.8:
reasons.append(("high_utilization", utilization))
# Check degradation metrics if provided
if metrics:
if metrics.get("attention_degradation", 0) > 0.3:
reasons.append(("attention_degradation", True))
if metrics.get("quality_score", 1.0) < 0.8:
reasons.append(("quality_degradation", True))
return len(reasons) > 0, reasons
# ---------------------------------------------------------------------------
# Cache optimization
# ---------------------------------------------------------------------------
def design_stable_prompt(template: str, dynamic_values: Optional[Dict] = None) -> str:
"""
Stabilize a prompt template for maximum KV-cache hit rate.
Use when: constructing system prompts or few-shot prefixes that will
be reused across many requests. Replaces dynamic content (timestamps,
session IDs, counters) with stable placeholders so the prefix hash
remains constant.
"""
result = template
# Replace timestamps
date_pattern = r'\d{4}-\d{2}-\d{2}'
result = re.sub(date_pattern, '[DATE_STABLE]', result)
# Replace session IDs
session_pattern = r'Session \d+'
result = re.sub(session_pattern, 'Session [STABLE]', result)
# Replace counters
counter_pattern = r'\d+/\d+'
result = re.sub(counter_pattern, '[COUNTER_STABLE]', result)
return result
def calculate_cache_metrics(
requests: List[Dict], cache: Dict[str, Dict]
) -> Dict[str, object]:
"""
Calculate KV-cache hit metrics for a request sequence.
Use when: evaluating whether prompt restructuring improved cache
utilization. Feed in the request log and current cache state to
get hit/miss rates and actionable recommendations.
"""
hits = 0
misses = 0
for req in requests:
prefix = req.get("prefix_hash", "")
token_count = req.get("token_count", 0)
if prefix in cache:
hits += token_count * cache[prefix].get("hit_ratio", 0)
else:
misses += token_count
total = hits + misses
return {
"hit_rate": hits / total if total > 0 else 0,
"cache_hits": hits,
"cache_misses": misses,
"recommendations": generate_cache_recommendations(hits, misses),
}
def generate_cache_recommendations(hits: int, misses: int) -> List[str]:
"""
Generate recommendations for cache optimization based on hit/miss ratio.
Use when: cache metrics indicate sub-optimal hit rates and concrete
next steps are needed.
"""
recommendations: List[str] = []
hit_rate = hits / (hits + misses) if (hits + misses) > 0 else 0
if hit_rate < 0.5:
recommendations.append("Consider stabilizing system prompts")
recommendations.append("Reduce variation in request prefixes")
if hit_rate < 0.8:
recommendations.append("Group similar requests together")
recommendations.append("Use consistent formatting across requests")
return recommendations
# ---------------------------------------------------------------------------
# Demo / smoke test
# ---------------------------------------------------------------------------
if __name__ == "__main__":
print("=== Context Optimization Utilities — Demo ===\n")
# 1. Token estimation
sample_text = "The quick brown fox jumps over the lazy dog. " * 20
tokens = estimate_token_count(sample_text)
print(f"1. Token estimate for {len(sample_text)}-char text: ~{tokens} tokens\n")
# 2. Observation masking
store = ObservationStore(max_size=100)
long_output = (
"Result: 42 items found\n"
"Total processing time: 3.2s\n"
"Details:\n" + "\n".join([f" Item {i}: value={i*10}" for i in range(20)])
)
masked, ref_id = store.mask(long_output, max_length=100)
print(f"2. Masked observation:\n {masked}")
print(f" Ref ID: {ref_id}")
retrieved = store.retrieve(ref_id)
print(f" Retrievable: {retrieved is not None}\n")
# 3. Context budget
budget = ContextBudget(total_limit=128_000)
budget.allocate("system_prompt", 1500)
budget.allocate("tool_definitions", 3000)
budget.allocate("message_history", 95_000)
usage = budget.get_usage()
print(f"3. Budget utilization: {usage['utilization_ratio']:.1%}")
should_opt, reasons = budget.should_optimize(
current_usage=int(128_000 * 0.85)
)
print(f" Should optimize: {should_opt}, reasons: {reasons}\n")
# 4. Cache-stable prompt
raw_prompt = "Session 42 started on 2025-12-20. Progress: 3/10 tasks."
stable = design_stable_prompt(raw_prompt)
print(f"4. Original prompt: {raw_prompt}")
print(f" Stabilized: {stable}\n")
# 5. Summarization
tool_out = "count: 150\nstatus: success\nFound 3 errors in module A."
summary = summarize_content(tool_out, "tool_output", max_length=200)
print(f"5. Tool output summary: {summary}\n")
print("=== Demo complete ===")