Agent Skills Framework Extension
Memory Context Patterns Skill
When to Use This Skill
Use this skill when implementing memory context patterns patterns in your codebase.
How to Use This Skill
- Review the patterns and examples below
- Apply the relevant patterns to your implementation
- Follow the best practices outlined in this skill
Context injection, memory retrieval, session continuity, and intelligent state management for AI systems.
Core Capabilities
- Context Injection - Intelligent context loading
- Memory Retrieval - Semantic and temporal search
- Session Continuity - Cross-session state preservation
- State Persistence - Durable storage patterns
- Knowledge Graph - Entity relationships
- Temporal Tracking - Time-aware context
Intelligent Context Injection
# scripts/context-injection.py
from dataclasses import dataclass
from typing import List, Dict, Optional, Any
from datetime import datetime, timedelta
import json
from pathlib import Path
@dataclass
class ContextChunk:
"""Single context chunk"""
id: str
content: str
source: str # 'session', 'knowledge', 'code', 'documentation'
timestamp: datetime
relevance_score: float
token_count: int
metadata: Dict[str, Any]
@dataclass
class ContextWindow:
"""Complete context window"""
chunks: List[ContextChunk]
total_tokens: int
max_tokens: int
def add_chunk(self, chunk: ContextChunk) -> bool:
"""Add chunk if it fits"""
if self.total_tokens + chunk.token_count > self.max_tokens:
return False
self.chunks.append(chunk)
self.total_tokens += chunk.token_count
return True
def remove_least_relevant(self):
"""Remove least relevant chunk"""
if not self.chunks:
return
least_relevant = min(self.chunks, key=lambda c: c.relevance_score)
self.chunks.remove(least_relevant)
self.total_tokens -= least_relevant.token_count
class ContextInjector:
"""Intelligently inject context into prompts"""
def __init__(self, max_context_tokens: int = 100000):
self.max_context_tokens = max_context_tokens
self.memory_db = self._load_memory_database()
def inject_context(
self,
prompt: str,
session_id: str,
project_context: Optional[str] = None
) -> str:
"""Build complete context for prompt"""
# Analyze prompt to understand needs
context_needs = self._analyze_prompt_needs(prompt)
# Build context window
window = ContextWindow(chunks=[], total_tokens=0, max_tokens=self.max_context_tokens)
# Priority 1: Recent session history
session_chunks = self._get_session_context(session_id, limit=10)
for chunk in session_chunks:
if not window.add_chunk(chunk):
window.remove_least_relevant()
window.add_chunk(chunk)
# Priority 2: Project context
if project_context:
project_chunks = self._get_project_context(project_context, context_needs)
for chunk in project_chunks:
if not window.add_chunk(chunk):
break # Stop if full
# Priority 3: Relevant knowledge
knowledge_chunks = self._search_knowledge(context_needs['keywords'], limit=5)
for chunk in knowledge_chunks:
if not window.add_chunk(chunk):
break
# Priority 4: Code examples
if context_needs['needs_code']:
code_chunks = self._get_code_examples(context_needs['technologies'])
for chunk in code_chunks[:3]:
if not window.add_chunk(chunk):
break
# Build final context
return self._format_context(window, prompt)
def _analyze_prompt_needs(self, prompt: str) -> Dict[str, Any]:
"""Analyze what context the prompt needs"""
needs = {
'keywords': [],
'technologies': [],
'needs_code': False,
'needs_history': False,
'temporal_scope': 'recent' # 'recent', 'all', 'specific'
}
prompt_lower = prompt.lower()
# Detect code needs
if any(word in prompt_lower for word in ['code', 'implement', 'function', 'class']):
needs['needs_code'] = True
# Detect technologies
tech_keywords = ['python', 'rust', 'react', 'typescript', 'sql', 'docker']
for tech in tech_keywords:
if tech in prompt_lower:
needs['technologies'].append(tech)
# Extract keywords (simple version)
words = prompt_lower.split()
needs['keywords'] = [w for w in words if len(w) > 4][:10]
return needs
def _get_session_context(self, session_id: str, limit: int) -> List[ContextChunk]:
"""Get recent session history"""
# Query session database
chunks = []
# Simulate database query
for i in range(min(limit, 5)):
chunks.append(ContextChunk(
id=f"session-{i}",
content=f"Previous interaction {i}",
source="session",
timestamp=datetime.now() - timedelta(minutes=i * 10),
relevance_score=0.9 - (i * 0.1),
token_count=100,
metadata={"session_id": session_id}
))
return chunks
def _get_project_context(self, project: str, needs: Dict) -> List[ContextChunk]:
"""Get relevant project context"""
chunks = []
# Load project README, architecture docs, etc.
project_path = Path(project)
if (project_path / "README.md").exists():
readme = (project_path / "README.md").read_text()
chunks.append(ContextChunk(
id="project-readme",
content=readme[:2000],
source="documentation",
timestamp=datetime.now(),
relevance_score=0.95,
token_count=len(readme.split()) * 1.3,
metadata={"file": "README.md"}
))
return chunks
def _search_knowledge(self, keywords: List[str], limit: int) -> List[ContextChunk]:
"""Search knowledge base"""
chunks = []
# Semantic search in knowledge database
# This would use embeddings in production
for i, keyword in enumerate(keywords[:limit]):
chunks.append(ContextChunk(
id=f"knowledge-{keyword}",
content=f"Knowledge about {keyword}",
source="knowledge",
timestamp=datetime.now() - timedelta(days=i),
relevance_score=0.8,
token_count=150,
metadata={"keyword": keyword}
))
return chunks
def _get_code_examples(self, technologies: List[str]) -> List[ContextChunk]:
"""Get relevant code examples"""
chunks = []
for tech in technologies:
chunks.append(ContextChunk(
id=f"code-{tech}",
content=f"# Example {tech} code\ndef example(): pass",
source="code",
timestamp=datetime.now(),
relevance_score=0.85,
token_count=50,
metadata={"technology": tech}
))
return chunks
def _format_context(self, window: ContextWindow, prompt: str) -> str:
"""Format context window with prompt"""
context_parts = []
# Group by source
by_source = {}
for chunk in window.chunks:
if chunk.source not in by_source:
by_source[chunk.source] = []
by_source[chunk.source].append(chunk)
# Format each source group
if 'session' in by_source:
context_parts.append("## Recent Session History\n")
for chunk in by_source['session']:
context_parts.append(f"{chunk.content}\n")
if 'documentation' in by_source:
context_parts.append("\n## Project Documentation\n")
for chunk in by_source['documentation']:
context_parts.append(f"{chunk.content}\n")
if 'knowledge' in by_source:
context_parts.append("\n## Relevant Knowledge\n")
for chunk in by_source['knowledge']:
context_parts.append(f"{chunk.content}\n")
if 'code' in by_source:
context_parts.append("\n## Code Examples\n")
for chunk in by_source['code']:
context_parts.append(f"```\n{chunk.content}\n```\n")
# Add the actual prompt
context_parts.append(f"\n## Current Request\n{prompt}\n")
return "\n".join(context_parts)
def _load_memory_database(self) -> Dict:
"""Load memory database"""
# In production: connect to SQLite/PostgreSQL
return {}
# Usage
injector = ContextInjector(max_context_tokens=100000)
prompt = "Implement a Python async function for API requests"
context = injector.inject_context(
prompt=prompt,
session_id="session-123",
project_context="/path/to/project"
)
print(context)
Memory Retrieval with Semantic Search
# scripts/memory-retrieval.py
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime, timedelta
import numpy as np
from pathlib import Path
import json
@dataclass
class MemoryEntry:
"""Single memory entry"""
id: str
content: str
embedding: np.ndarray
timestamp: datetime
category: str
tags: List[str]
access_count: int = 0
last_accessed: Optional[datetime] = None
class SemanticMemory:
"""Semantic memory with embedding-based retrieval"""
def __init__(self, storage_path: Path):
self.storage_path = storage_path
self.entries: List[MemoryEntry] = []
self.load()
def store(self, content: str, category: str, tags: List[str]):
"""Store new memory"""
# Generate embedding (simplified)
embedding = self._generate_embedding(content)
entry = MemoryEntry(
id=f"mem-{len(self.entries)}",
content=content,
embedding=embedding,
timestamp=datetime.now(),
category=category,
tags=tags
)
self.entries.append(entry)
self.save()
def retrieve(
self,
query: str,
limit: int = 5,
recency_weight: float = 0.3
) -> List[MemoryEntry]:
"""Retrieve relevant memories"""
query_embedding = self._generate_embedding(query)
# Calculate similarity scores
scores = []
for entry in self.entries:
# Semantic similarity
semantic_sim = self._cosine_similarity(query_embedding, entry.embedding)
# Recency score (newer = higher)
age_hours = (datetime.now() - entry.timestamp).total_seconds() / 3600
recency_score = 1.0 / (1.0 + age_hours / 24) # Decay over days
# Combined score
combined = (
semantic_sim * (1 - recency_weight) +
recency_score * recency_weight
)
scores.append((combined, entry))
# Sort by score
scores.sort(reverse=True, key=lambda x: x[0])
# Update access stats
results = []
for score, entry in scores[:limit]:
entry.access_count += 1
entry.last_accessed = datetime.now()
results.append(entry)
self.save()
return results
def retrieve_temporal(
self,
query: str,
time_range: timedelta,
limit: int = 5
) -> List[MemoryEntry]:
"""Retrieve memories within time range"""
cutoff = datetime.now() - time_range
# Filter by time
recent = [e for e in self.entries if e.timestamp >= cutoff]
# Semantic search within recent
query_embedding = self._generate_embedding(query)
scores = [
(self._cosine_similarity(query_embedding, e.embedding), e)
for e in recent
]
scores.sort(reverse=True, key=lambda x: x[0])
return [e for _, e in scores[:limit]]
def _generate_embedding(self, text: str) -> np.ndarray:
"""Generate text embedding (simplified)"""
# In production: use sentence-transformers or OpenAI embeddings
# For now: simple bag-of-words hash
words = text.lower().split()
embedding = np.zeros(128)
for i, word in enumerate(words[:128]):
embedding[hash(word) % 128] = 1.0
# Normalize
norm = np.linalg.norm(embedding)
if norm > 0:
embedding /= norm
return embedding
def _cosine_similarity(self, a: np.ndarray, b: np.ndarray) -> float:
"""Calculate cosine similarity"""
return float(np.dot(a, b))
def save(self):
"""Save memory to disk"""
self.storage_path.mkdir(parents=True, exist_ok=True)
data = []
for entry in self.entries:
data.append({
'id': entry.id,
'content': entry.content,
'embedding': entry.embedding.tolist(),
'timestamp': entry.timestamp.isoformat(),
'category': entry.category,
'tags': entry.tags,
'access_count': entry.access_count,
'last_accessed': entry.last_accessed.isoformat() if entry.last_accessed else None
})
(self.storage_path / 'memory.json').write_text(json.dumps(data, indent=2))
def load(self):
"""Load memory from disk"""
memory_file = self.storage_path / 'memory.json'
if not memory_file.exists():
return
data = json.loads(memory_file.read_text())
self.entries = [
MemoryEntry(
id=e['id'],
content=e['content'],
embedding=np.array(e['embedding']),
timestamp=datetime.fromisoformat(e['timestamp']),
category=e['category'],
tags=e['tags'],
access_count=e['access_count'],
last_accessed=datetime.fromisoformat(e['last_accessed']) if e['last_accessed'] else None
)
for e in data
]
# Usage
memory = SemanticMemory(Path('.coditect/memory'))
# Store
memory.store(
"Implemented async API client using aiohttp",
category="implementation",
tags=["python", "async", "api"]
)
# Retrieve
results = memory.retrieve("How to make async HTTP requests?", limit=3)
for result in results:
print(f"[{result.timestamp}] {result.content}")
Session Continuity Manager
// scripts/session-continuity.ts
interface SessionState {
sessionId: string;
startTime: Date;
lastActivity: Date;
context: Map<string, any>;
checkpoints: Checkpoint[];
}
interface Checkpoint {
timestamp: Date;
state: Record<string, any>;
description: string;
}
class SessionContinuityManager {
private sessions = new Map<string, SessionState>();
private storagePath: string;
constructor(storagePath: string) {
this.storagePath = storagePath;
this.loadSessions();
}
/**
* Create or resume session
*/
getOrCreateSession(sessionId: string): SessionState {
if (this.sessions.has(sessionId)) {
const session = this.sessions.get(sessionId)!;
session.lastActivity = new Date();
return session;
}
const newSession: SessionState = {
sessionId,
startTime: new Date(),
lastActivity: new Date(),
context: new Map(),
checkpoints: [],
};
this.sessions.set(sessionId, newSession);
return newSession;
}
/**
* Create checkpoint
*/
createCheckpoint(
sessionId: string,
description: string,
state: Record<string, any>
): void {
const session = this.getOrCreateSession(sessionId);
session.checkpoints.push({
timestamp: new Date(),
state: { ...state },
description,
});
this.saveSessions();
}
/**
* Restore from checkpoint
*/
restoreCheckpoint(sessionId: string, index: number = -1): Record<string, any> | null {
const session = this.sessions.get(sessionId);
if (!session || session.checkpoints.length === 0) {
return null;
}
const checkpointIndex = index < 0
? session.checkpoints.length + index
: index;
if (checkpointIndex < 0 || checkpointIndex >= session.checkpoints.length) {
return null;
}
return session.checkpoints[checkpointIndex].state;
}
/**
* Get session context
*/
getContext(sessionId: string, key: string): any {
const session = this.sessions.get(sessionId);
return session?.context.get(key);
}
/**
* Set session context
*/
setContext(sessionId: string, key: string, value: any): void {
const session = this.getOrCreateSession(sessionId);
session.context.set(key, value);
this.saveSessions();
}
/**
* Export session for continuity
*/
exportSession(sessionId: string): string {
const session = this.sessions.get(sessionId);
if (!session) {
throw new Error(`Session ${sessionId} not found`);
}
const exported = {
sessionId: session.sessionId,
startTime: session.startTime.toISOString(),
lastActivity: session.lastActivity.toISOString(),
context: Object.fromEntries(session.context),
checkpoints: session.checkpoints.map(cp => ({
timestamp: cp.timestamp.toISOString(),
state: cp.state,
description: cp.description,
})),
};
return JSON.stringify(exported, null, 2);
}
/**
* Import session
*/
importSession(data: string): void {
const imported = JSON.parse(data);
const session: SessionState = {
sessionId: imported.sessionId,
startTime: new Date(imported.startTime),
lastActivity: new Date(imported.lastActivity),
context: new Map(Object.entries(imported.context)),
checkpoints: imported.checkpoints.map((cp: any) => ({
timestamp: new Date(cp.timestamp),
state: cp.state,
description: cp.description,
})),
};
this.sessions.set(session.sessionId, session);
this.saveSessions();
}
private saveSessions(): void {
// Save to file system
const data = Array.from(this.sessions.values()).map(session => ({
sessionId: session.sessionId,
startTime: session.startTime.toISOString(),
lastActivity: session.lastActivity.toISOString(),
context: Object.fromEntries(session.context),
checkpoints: session.checkpoints.map(cp => ({
timestamp: cp.timestamp.toISOString(),
state: cp.state,
description: cp.description,
})),
}));
// In practice: write to file
console.log('Saving sessions...', data.length);
}
private loadSessions(): void {
// Load from file system
// In practice: read from file and restore sessions
}
}
// Usage
const manager = new SessionContinuityManager('.coditect/sessions');
// Create session
const session = manager.getOrCreateSession('session-123');
// Set context
manager.setContext('session-123', 'currentProject', 'my-app');
manager.setContext('session-123', 'lastCommand', '/git-sync');
// Create checkpoint
manager.createCheckpoint('session-123', 'Before major refactor', {
files: ['src/main.ts', 'src/utils.ts'],
branch: 'feature/refactor',
});
// Later: restore checkpoint
const state = manager.restoreCheckpoint('session-123', -1);
console.log('Restored state:', state);
// Export for continuity
const exported = manager.exportSession('session-123');
console.log(exported);
Usage Examples
Intelligent Context Injection
Apply memory-context-patterns skill to inject relevant context for prompt with project and session history
Semantic Memory Retrieval
Apply memory-context-patterns skill to retrieve relevant memories using semantic search with recency weighting
Session Continuity
Apply memory-context-patterns skill to create session checkpoint and export for cross-session continuity
Success Output
When successful, this skill MUST output:
✅ SKILL COMPLETE: memory-context-patterns
Completed:
- [x] Context injection system implemented with priority-based loading
- [x] Semantic memory retrieval functional with cosine similarity search
- [x] Session continuity manager storing and restoring checkpoints
- [x] Context window management respecting 100K token limit
- [x] Multi-source context aggregation (session, project, knowledge, code)
Outputs:
- scripts/context-injection.py (intelligent context loading system)
- scripts/memory-retrieval.py (semantic search with embeddings)
- scripts/session-continuity.ts (checkpoint management)
- .coditect/memory/memory.json (semantic memory database)
- .coditect/sessions/session-{id}.json (session state storage)
Performance:
- Context injection: <2s for typical prompt with 5K token context
- Semantic search: <500ms for 1000 memory entries
- Session restore: <1s from checkpoint
- Memory storage: <10MB for 1000 entries
Completion Checklist
Before marking this skill as complete, verify:
- Context injector loads session history from last 10 interactions
- Semantic search retrieves relevant memories based on query
- Session checkpoints restore previous state accurately
- Context window never exceeds configured token limit
- Memory entries include embeddings for similarity search
- Session export/import works across context windows
- All Python/TypeScript dependencies installed (numpy, datetime)
- Integration tests pass for context injection pipeline
Failure Indicators
This skill has FAILED if:
- ❌ Context injection exceeds token limit causing truncation errors
- ❌ Semantic search returns irrelevant memories (cosine similarity <0.3)
- ❌ Session checkpoints fail to restore or restore incorrect state
- ❌ Context window management doesn't prioritize recent/relevant content
- ❌ Memory database grows unbounded without cleanup strategy
- ❌ Session export format incompatible across different sessions
- ❌ Context loading takes >5s, blocking prompt execution
When NOT to Use
Do NOT use memory-context-patterns when:
- Working with stateless single-turn interactions (no context needed)
- Context requirements fit within 10K tokens (use simple string concatenation)
- Using external vector database (Pinecone, Weaviate) for memory (different pattern)
- Session persistence not required (ephemeral conversations)
- Semantic search not needed (exact keyword matching sufficient)
- Real-time constraints require <100ms latency (this adds 500ms-2s overhead)
- Memory requirements exceed local storage capacity (use cloud memory service)
Anti-Patterns (Avoid)
| Anti-Pattern | Problem | Solution |
|---|---|---|
| Loading entire history | Token limit exceeded, slow | Use priority-based windowing with recency decay |
| No embedding normalization | Incorrect similarity scores | Normalize embeddings to unit vectors before comparison |
| Synchronous context loading | Blocks prompt execution | Load context asynchronously or use caching |
| No memory expiration | Database grows unbounded | Implement TTL-based cleanup or LRU eviction |
| Hardcoded token limits | Breaks with model changes | Read token limit from model config |
| No context deduplication | Redundant information | Hash content and dedupe before injection |
| Storing raw text only | No semantic search capability | Generate and store embeddings alongside text |
Principles
This skill embodies:
- #5 Eliminate Ambiguity - Context windowing rules clearly prioritize recent and relevant
- #6 Clear, Understandable, Explainable - Context sources labeled (session, project, knowledge)
- #8 No Assumptions - Token limits enforced, context needs analyzed from prompt
- #9 Quality Over Speed - Semantic search accuracy prioritized over retrieval speed
- #12 Separation of Concerns - Context injection separate from memory storage
Full Principles: CODITECT-STANDARD-AUTOMATION.md
Integration Points
- memory-optimization-patterns - Token reduction strategies
- session-analysis-patterns - Session insight extraction
- prompt-analysis-patterns - Context need analysis