原始内容
name: smart-context-manager description: Intelligent context management. Solves context decay problem identified in research papers, implements memory hierarchy (short-term/medium-term/long-term), supports decay detection, intelligent compression, and retrieval augmentation. Optimizes performance and accuracy in long-session scenarios. version: 1.0.0
Smart Context Manager
Academic Foundation
This skill is based on the findings from the comprehensive survey paper:
"Agent Harness Engineering: A Survey" (2026)
Authors: Junjie Li (CMU), Xi Xiao (UAB), Yunbei Zhang (Tulane), Chen Liu (Yale), Lin Zhao (NEU), Xiaoying Liao (JHU), Yingrui Ji (UAB), Janet Wang (UAB), Jianyang Gu (OSU), Yingqiang Ge (Amazon), Weijie Xu (Amazon), Xi Fang (Amazon), Xiang Xu (Amazon), Tianchen Zhao (Amazon), Youngeun Kim (Amazon), Tianyang Wang (UAB), Jihun Hamm (Tulane), Smita Krishnaswamy (Yale), Jun Huan (Amazon), Chandan K Reddy (Virginia Tech & Amazon)
Institutions: CMU, Yale, JHU, NEU, Tulane, UAB, OSU, Virginia Tech, Amazon
Project Page: Awesome-Agent-Harness
This survey analyzed 170+ open-source projects and distilled engineering principles from production deployments at OpenAI, Anthropic, and LangChain.
Core Philosophy
From the "Agent Harness Engineering" survey paper:
"Larger context windows do not solve the memory problem. There exists a U-shaped attention curve (accuracy drops 30%+ when information is in the middle) and context decay phenomenon."
Key Paradigm Shifts:
- From "stuffing context" to "intelligent layered management"
- From "passive compression" to "proactive decay detection"
- From "full memory" to "retrieval augmentation"
- From "single storage" to "multi-level memory"
Key Research Findings
1. U-Shaped Attention Curve
Accuracy
^
| ╱╲
| ╱ ╲
| ╱ ╲
| ╱ ╲
| ╱ ╲
|╱ ╲
+-----------------> Information Position
Start Middle End
Key Findings:
- Information at the beginning: High accuracy
- Information in the middle: Accuracy drops 30%+
- Information at the end: High accuracy
Implication: Critical information should be placed at the beginning or end, avoiding the middle.
2. Context Decay
Memory Strength
^
|╲
| ╲
| ╲
| ╲
| ╲
| ╲
+-----------> Time/Conversation Rounds
Key Findings:
- Early information gradually gets "forgotten"
- Even within the context window, it gets ignored
- Need to proactively refresh critical information
Implication: Need to periodically restate key information, rather than assuming the model will remember.
3. Medium-Term Memory Gap
Memory Hierarchy:
├─ Short-term Memory: Current context window (existing)
├─ Medium-term Memory: Session-level persistence (missing)
└─ Long-term Memory: Cross-session storage (MEMORY.md)
Implication: Need to establish session-level medium-term memory to fill the gap between short-term and long-term.
Memory Hierarchy Structure
Level 1: Working Memory
Definition: Active information currently being processed.
Characteristics:
Capacity: Limited (constrained by context window)
Duration: Current task
Access Speed: Instant
Update Frequency: High
Stored Content:
- Current task description
- Recent conversation rounds
- Active variables and states
- Pending issues
Management Strategies:
- Keep concise: Only retain necessary information
- Position optimization: Place critical info at beginning/end
- Regular cleanup: Remove no longer needed information
Level 2: Short-term Memory
Definition: Contextual information of the current session.
Characteristics:
Capacity: Medium (current session)
Duration: Current session
Access Speed: Fast (within context window)
Update Frequency: Medium
Stored Content:
- Conversation history from session start
- Completed task records
- Temporary decisions and preferences
- Intermediate results and states
Management Strategies:
- Rolling window: Keep most recent N rounds
- Intelligent compression: Compress earlier conversations
- Key extraction: Extract important info to medium-term memory
Level 3: Medium-term Memory
Definition: Structured information persisted at session level.
Characteristics:
Capacity: Large
Duration: Current session, recoverable across sessions
Access Speed: Medium (requires retrieval)
Update Frequency: Low
Stored Content:
- Session goals and outline
- Key decisions and rationale
- Important findings and insights
- User preferences and constraints
- Task progress and status
Management Strategies:
- Structured storage: Use explicit format
- Categorical management: Organize by type
- Index acceleration: Build fast retrieval index
- Regular updates: Update after each significant operation
Level 4: Long-term Memory
Definition: Knowledge persisted across sessions.
Characteristics:
Capacity: Unlimited
Duration: Permanent
Access Speed: Slow (requires retrieval)
Update Frequency: Very low
Stored Content:
- User profile and preferences
- Project information and configuration
- Accumulated knowledge and experience
- Important historical events
Management Strategies:
- Curated storage: Only keep high-value information
- Periodic review: Regularly review and update
- Categorized indexing: Facilitate retrieval
- Version control: Record change history
Core Features
1. Decay Detector
Purpose: Detect whether information is decaying and needs refresh.
Implementation:
# scripts/decay_detector.py
class DecayDetector:
"""Context Decay Detector"""
def __init__(self, decay_threshold: int = 20):
"""
Args:
decay_threshold: Number of rounds threshold for info to be considered "decayed"
"""
self.decay_threshold = decay_threshold
self.information_records = {} # Records each info's position and last reference
def register_info(self, info_id: str, info_type: str, content: str, position: int):
"""Register a piece of information"""
self.information_records[info_id] = {
"type": info_type,
"content": content,
"position": position, # Position in context (0=beginning)
"last_referenced": 0, # Last referenced round
"reference_count": 0, # Total reference count
"importance": self._calculate_importance(info_type, content)
}
def record_reference(self, info_id: str, current_round: int):
"""Record information being referenced"""
if info_id in self.information_records:
self.information_records[info_id]["last_referenced"] = current_round
self.information_records[info_id]["reference_count"] += 1
def detect_decayed_info(self, current_round: int) -> List[Dict]:
"""Detect decayed information"""
decayed = []
for info_id, record in self.information_records.items():
rounds_since_ref = current_round - record["last_referenced"]
# Decay conditions
is_decayed = (
rounds_since_ref >= self.decay_threshold or
(record["importance"] == "high" and rounds_since_ref >= self.decay_threshold // 2)
)
if is_decayed:
decayed.append({
"id": info_id,
"type": record["type"],
"content": record["content"],
"rounds_since_reference": rounds_since_ref,
"importance": record["importance"],
"recommendation": self._get_refresh_recommendation(record)
})
return decayed
def _calculate_importance(self, info_type: str, content: str) -> str:
"""Calculate information importance"""
high_importance_types = ["user_preference", "constraint", "goal", "critical_info"]
medium_importance_types = ["decision", "state", "progress"]
if info_type in high_importance_types:
return "high"
elif info_type in medium_importance_types:
return "medium"
else:
return "low"
def _get_refresh_recommendation(self, record: Dict) -> str:
"""Get refresh recommendation"""
if record["importance"] == "high":
return "Restate immediately: High importance information is decaying"
elif record["importance"] == "medium":
return "Consider restating: Medium importance information has decayed"
else:
return "Can ignore: Low importance information, no refresh needed"
def get_position_warning(self, current_position: int, total_length: int) -> Optional[str]:
"""Check if position is in danger zone (U-curve middle)"""
# Middle 40% region is danger zone
danger_start = total_length * 0.3
danger_end = total_length * 0.7
if danger_start <= current_position <= danger_end:
return (
f"⚠️ Position Warning: Information is in context middle zone "
f"({current_position}/{total_length}), may be ignored. "
f"Suggestion: Move to beginning or end."
)
return None
Output Example:
Decay Detection Report
================================
Current Round: 45
Decayed Information:
[High Importance] user_preference: Use Feishu notifications
Last referenced: 18 rounds ago
Suggestion: Restate immediately: High importance information is decaying
[Medium Importance] constraint: Do not delete files
Last referenced: 25 rounds ago
Suggestion: Consider restating: Medium importance information has decayed
Position Warnings:
Information "Project Progress" is in danger zone (position 1500/5000)
Suggestion: Move to beginning or end to improve attention
2. Smart Compressor
Purpose: Compress context while preserving key information.
Implementation:
# scripts/smart_compressor.py
class SmartCompressor:
"""Intelligent Context Compressor"""
def __init__(self):
self.compression_strategies = {
"conversation": self._compress_conversation,
"code": self._compress_code,
"decision": self._compress_decision,
"error": self._compress_error
}
def compress(self, context: Dict, target_ratio: float = 0.3) -> Dict:
"""
Compress context
Args:
context: Original context
target_ratio: Target compression ratio (compressed/original)
Returns:
Compressed context
"""
original_size = self._estimate_size(context)
target_size = int(original_size * target_ratio)
# Categorize content
categorized = self._categorize_content(context)
# Prioritize by importance
prioritized = self._prioritize_content(categorized)
# Select content to retain
selected = self._select_content(prioritized, target_size)
# Compress each category
compressed = {}
for category, items in selected.items():
strategy = self.compression_strategies.get(category, self._compress_generic)
compressed[category] = strategy(items)
return compressed
def _categorize_content(self, context: Dict) -> Dict[str, List]:
"""Categorize content"""
categorized = {
"conversation": [],
"code": [],
"decision": [],
"error": []
}
for item in context.get("history", []):
if item.get("type") == "user_message" or item.get("type") == "assistant_message":
categorized["conversation"].append(item)
elif item.get("type") == "code":
categorized["code"].append(item)
elif item.get("type") == "decision":
categorized["decision"].append(item)
elif item.get("type") == "error":
categorized["error"].append(item)
return categorized
def _prioritize_content(self, categorized: Dict) -> List:
"""Prioritize by priority"""
priorities = {
"decision": 100, # Decisions most important
"error": 90, # Errors next (for debugging)
"code": 70, # Code medium
"conversation": 50 # Conversation lowest
}
prioritized = []
for category, items in categorized.items():
priority = priorities.get(category, 50)
for item in items:
# Adjust priority
adjusted_priority = priority + self._adjust_priority(item)
prioritized.append((adjusted_priority, category, item))
return sorted(prioritized, key=lambda x: -x[0])
def _adjust_priority(self, item: Dict) -> int:
"""Adjust priority"""
adjustment = 0
# Recent information has higher priority
if item.get("recency") == "recent":
adjustment += 20
# Important-marked information has higher priority
if item.get("important"):
adjustment += 30
# Successful operations have higher priority
if item.get("status") == "success":
adjustment += 10
return adjustment
def _select_content(self, prioritized: List, target_size: int) -> Dict:
"""Select content to retain"""
selected = {}
current_size = 0
for priority, category, item in prioritized:
item_size = self._estimate_item_size(item)
if current_size + item_size <= target_size:
if category not in selected:
selected[category] = []
selected[category].append(item)
current_size += item_size
return selected
def _compress_conversation(self, items: List) -> List:
"""Compress conversations"""
compressed = []
for item in items:
# Keep summary instead of full content
compressed.append({
"role": item.get("role"),
"summary": self._summarize_text(item.get("content", "")),
"key_points": self._extract_key_points(item.get("content", "")),
"timestamp": item.get("timestamp")
})
return compressed
def _compress_code(self, items: List) -> List:
"""Compress code"""
compressed = []
for item in items:
# Keep code structure and key parts
compressed.append({
"file": item.get("file"),
"summary": f"{item.get('lines_count', 0)} lines",
"key_functions": self._extract_function_names(item.get("content", "")),
"changes": item.get("changes", [])
})
return compressed
def _compress_decision(self, items: List) -> List:
"""Compress decisions (keep complete)"""
return items # Decisions are important, try to keep complete
def _compress_error(self, items: List) -> List:
"""Compress errors"""
compressed = []
for item in items:
# Keep error summary and resolution
compressed.append({
"error_type": item.get("error_type"),
"message": item.get("message")[:200], # Truncate long error messages
"solution": item.get("solution"),
"resolved": item.get("resolved")
})
return compressed
def _compress_generic(self, items: List) -> List:
"""Generic compression"""
return [{"summary": str(item)[:200]} for item in items]
def _estimate_size(self, context: Dict) -> int:
"""Estimate context size (token count)"""
# Simplified estimation: character count / 4
text = json.dumps(context, ensure_ascii=False)
return len(text) // 4
def _estimate_item_size(self, item: Dict) -> int:
"""Estimate single item size"""
text = json.dumps(item, ensure_ascii=False)
return len(text) // 4
def _summarize_text(self, text: str, max_length: int = 100) -> str:
"""Summarize text"""
if len(text) <= max_length:
return text
return text[:max_length] + "..."
def _extract_key_points(self, text: str) -> List[str]:
"""Extract key points"""
# Simplified implementation: extract keywords
keywords = []
# In practice, can use NLP techniques
return keywords
def _extract_function_names(self, code: str) -> List[str]:
"""Extract function names"""
import re
pattern = r'def\s+(\w+)\s*\('
return re.findall(pattern, code)
3. Retrieval Enhancer
Purpose: Retrieve relevant information from memory to enhance current context.
Implementation:
# scripts/retrieval_enhancer.py
class RetrievalEnhancer:
"""Retrieval Enhancer"""
def __init__(self, memory_paths: Dict[str, Path]):
"""
Args:
memory_paths: Paths to each memory level
"""
self.memory_paths = memory_paths
self.index = {} # Retrieval index
def build_index(self):
"""Build retrieval index"""
for level, path in self.memory_paths.items():
if path.exists():
content = path.read_text(encoding="utf-8")
self.index[level] = self._parse_content(content)
def retrieve(self, query: str, top_k: int = 5) -> List[Dict]:
"""
Retrieve relevant information
Args:
query: Query text
top_k: Return top K results
Returns:
List of retrieval results
"""
results = []
for level, entries in self.index.items():
for entry in entries:
score = self._calculate_relevance(query, entry)
if score > 0.5: # Relevance threshold
results.append({
"level": level,
"content": entry["content"],
"score": score,
"metadata": entry.get("metadata", {})
})
# Sort by relevance
results.sort(key=lambda x: -x["score"])
return results[:top_k]
def enhance_context(self, current_context: str, query: str) -> str:
"""
Enhance current context
Args:
current_context: Current context
query: Current query
Returns:
Enhanced context
"""
# Retrieve relevant information
relevant_info = self.retrieve(query)
if not relevant_info:
return current_context
# Build enhancement section
enhancement = "\n\n## Relevant Context (Retrieval Augmented)\n\n"
for info in relevant_info:
enhancement += f"**[{info['level']}]** {info['content'][:200]}\n\n"
# Place enhancement at beginning (U-curve optimization)
return enhancement + current_context
def _parse_content(self, content: str) -> List[Dict]:
"""Parse content into entry list"""
# Simplified implementation: split by paragraph
paragraphs = content.split("\n\n")
entries = []
for para in paragraphs:
if para.strip():
entries.append({
"content": para.strip(),
"metadata": {}
})
return entries
def _calculate_relevance(self, query: str, entry: Dict) -> float:
"""Calculate relevance"""
# Simplified implementation: keyword matching
query_words = set(query.lower().split())
content_words = set(entry["content"].lower().split())
if not query_words or not content_words:
return 0.0
intersection = query_words & content_words
union = query_words | content_words
return len(intersection) / len(union) if union else 0.0
4. Memory Synchronizer
Purpose: Synchronize memory across levels to ensure consistency.
Implementation:
# scripts/memory_sync.py
class MemorySynchronizer:
"""Memory Synchronizer"""
def __init__(self, session_file: Path, long_term_file: Path):
self.session_file = session_file
self.long_term_file = long_term_file
def promote_to_long_term(self, info: Dict) -> bool:
"""
Promote important information to long-term memory
Conditions:
- High importance
- Cross-session value
- User confirmation
"""
if not self._should_promote(info):
return False
# Read long-term memory
long_term = self._read_long_term()
# Add new entry
long_term["entries"].append({
"content": info["content"],
"source": "session",
"promoted_at": datetime.now().isoformat(),
"importance": info["importance"]
})
# Write back
self._write_long_term(long_term)
return True
def sync_session_to_long_term(self):
"""Sync session memory to long-term memory"""
session = self._read_session()
long_term = self._read_long_term()
# Find content needing sync
for entry in session.get("pending_promotion", []):
if self._should_promote(entry):
long_term["entries"].append(entry)
# Write back
self._write_long_term(long_term)
# Clear pending promotion list in session
session["pending_promotion"] = []
self._write_session(session)
def _should_promote(self, info: Dict) -> bool:
"""Determine whether to promote"""
# High importance
if info.get("importance") == "high":
return True
# User preference
if info.get("type") == "user_preference":
return True
# Project configuration
if info.get("type") == "project_config":
return True
return False
def _read_session(self) -> Dict:
"""Read session memory"""
if self.session_file.exists():
return json.loads(self.session_file.read_text(encoding="utf-8"))
return {"entries": [], "pending_promotion": []}
def _read_long_term(self) -> Dict:
"""Read long-term memory"""
if self.long_term_file.exists():
content = self.long_term_file.read_text(encoding="utf-8")
# Parse MEMORY.md format
return {"entries": self._parse_memory_md(content)}
return {"entries": []}
def _write_session(self, data: Dict):
"""Write session memory"""
self.session_file.write_text(
json.dumps(data, ensure_ascii=False, indent=2),
encoding="utf-8"
)
def _write_long_term(self, data: Dict):
"""Write long-term memory"""
# Convert to MEMORY.md format
content = self._format_memory_md(data)
self.long_term_file.write_text(content, encoding="utf-8")
def _parse_memory_md(self, content: str) -> List[Dict]:
"""Parse MEMORY.md format"""
# Simplified implementation
return []
def _format_memory_md(self, data: Dict) -> str:
"""Format as MEMORY.md"""
lines = ["# Session Memory Sync\n"]
for entry in data.get("entries", []):
lines.append(f"- {entry.get('content', '')}\n")
return "".join(lines)
Usage Scenarios
Scenario 1: Long Session Management
# Detect decay
python scripts/context_manager.py detect-decay --current-round 45
# Smart compression
python scripts/context_manager.py compress --target-ratio 0.3
# Retrieval augmentation
python scripts/context_manager.py enhance --query "What are user preferences"
Scenario 2: Critical Information Protection
# Register critical information
python scripts/context_manager.py register --id "user_pref" --type "user_preference" --content "Use Feishu notifications"
# Check position
python scripts/context_manager.py check-position --id "user_pref" --position 2500 --total 5000
# Refresh decayed information
python scripts/context_manager.py refresh --id "user_pref"
Scenario 3: Session End Synchronization
# Sync to long-term memory
python scripts/context_manager.py sync
# Generate compressed summary
python scripts/context_manager.py summarize-session
Integration with Existing Skills
Integration with compact
Enhanced compact skill:
Original:
Compress entire conversation history → Summary
Enhanced:
1. Detect decayed information → Immediately restate
2. Categorize content → Compress by type
3. Preserve key decisions → Keep complete
4. Compress conversation → Summary
5. Sync to medium-term memory → Persist
Integration with context-monitor
Enhanced context-monitor:
Original:
Monitor context size → Threshold alert
Enhanced:
1. Monitor context size
2. Detect information decay (NEW)
3. Detect position risk (NEW)
4. Smart compression suggestions (NEW)
5. Auto-refresh critical information (NEW)
Integration with agent-observability-suite
Observability Support:
Traces:
- Each memory retrieval
- Compression operations
- Information refresh
Metrics:
- Context size trend
- Decay event count
- Retrieval hit rate
Audit:
- Critical information registration
- Long-term memory promotion
File Structure
smart-context-manager/
├── SKILL.md # This file
├── scripts/
│ ├── context_manager.py # Main script
│ ├── decay_detector.py # Decay detection
│ ├── smart_compressor.py # Smart compression
│ ├── retrieval_enhancer.py # Retrieval augmentation
│ └── memory_sync.py # Memory synchronization
├── templates/
│ └── memory_structure.yaml # Memory structure template
└── references/
├── u_curve_analysis.md # U-curve analysis
└── memory_hierarchy.md # Memory hierarchy details
Configuration Example
# .dumate/context/config.yaml
context_manager:
# Decay detection config
decay:
enabled: true
threshold_rounds: 20 # Default decay threshold
high_importance_threshold: 10 # High importance info threshold
auto_refresh: true # Auto-refresh decayed information
# Compression config
compression:
enabled: true
target_ratio: 0.3 # Compression target ratio
preserve_types: # Types to keep complete
- decision
- constraint
- user_preference
# Retrieval augmentation config
retrieval:
enabled: true
top_k: 5 # Return top K results
relevance_threshold: 0.5 # Relevance threshold
enhance_position: "start" # Enhancement position (start/end)
# Memory sync config
sync:
enabled: true
auto_promote: true # Auto-promote high-value information
promotion_threshold: # Promotion threshold
importance: high
cross_session_value: true
# Position optimization config
position:
enabled: true
danger_zone: [0.3, 0.7] # Danger zone (middle 40%)
auto_relocate: true # Auto-relocate
Best Practices
1. Information Layering
Divide information into four levels, choose storage location by importance:
- Working Memory: Critical information for current task
- Short-term Memory: Session-level temporary information
- Medium-term Memory: Structured session summary
- Long-term Memory: High-value persistent information
2. Position Optimization
Leverage U-shaped attention curve:
- Beginning: Key constraints, user preferences, task goals
- Middle: Detailed processes, historical records
- End: Pending issues, next actions
3. Proactive Refreshing
Don't assume the model will remember:
- Periodically restate critical information
- Proactively refresh after decay detection
- Re-explain important decisions when needed
4. Smart Compression
Compression is not simple truncation:
- Preserve decisions and rationale
- Summarize processes rather than delete
- Maintain completeness of key information
5. Retrieval Augmentation
When context is insufficient, actively retrieve:
- Retrieve from medium/long-term memory
- Place retrieval results at beginning
- Avoid duplicate information
Bottom Line
Larger context windows do not equal better memory.
Information persistence requires proactive management, not passive storage.
Decay detection is more important than after-the-fact compression.