20 KiB
BMAD Context Management Engine
Overview
The Context Management Engine provides sophisticated context preservation, sharing, and persistence across persona interactions, ensuring seamless continuity and intelligent context-aware decision making throughout the BMAD Method workflow.
Core Architecture
Context Data Model
Hierarchical Context Structure
```yaml context_hierarchy: global_context: description: "System-wide context shared across all interactions" components: - "user_preferences" - "system_configuration" - "global_constraints" - "organizational_standards"
project_context: description: "Project-specific context shared within project scope" components: - "project_goals_and_objectives" - "stakeholder_information" - "technical_requirements" - "timeline_and_milestones" - "budget_and_resource_constraints" - "quality_standards"
session_context: description: "Session-specific context for current interaction" components: - "current_conversation_history" - "active_personas" - "workflow_state" - "pending_decisions" - "temporary_artifacts"
persona_context: description: "Persona-specific context and working memory" components: - "persona_working_memory" - "specialized_knowledge_cache" - "persona_specific_preferences" - "task_specific_context" - "collaboration_history"
#### Context Entity Schema
\```json
{
"context_entity": {
"id": "unique_context_identifier",
"type": "global|project|session|persona",
"scope": "context_scope_definition",
"created_at": "timestamp",
"updated_at": "timestamp",
"version": "context_version_number",
"metadata": {
"owner": "context_owner",
"access_level": "public|private|restricted",
"retention_policy": "retention_duration",
"encryption_level": "none|standard|high"
},
"content": {
"structured_data": {},
"unstructured_data": "",
"relationships": [],
"tags": [],
"priority": "high|medium|low"
},
"access_control": {
"read_permissions": [],
"write_permissions": [],
"share_permissions": []
}
}
}
Context Lifecycle Management
Context Creation and Initialization
def create_context(context_type, scope, initial_data, metadata=None):
"""
Create new context entity with proper initialization
"""
context_id = generate_context_id(context_type, scope)
context_entity = {
'id': context_id,
'type': context_type,
'scope': scope,
'created_at': get_current_timestamp(),
'updated_at': get_current_timestamp(),
'version': 1,
'metadata': initialize_metadata(metadata),
'content': structure_initial_data(initial_data),
'access_control': initialize_access_control(context_type)
}
# Validate context structure
validate_context_structure(context_entity)
# Store context
store_context(context_entity)
# Initialize relationships
initialize_context_relationships(context_entity)
return context_id
def initialize_metadata(metadata):
"""Initialize context metadata with defaults"""
default_metadata = {
'owner': get_current_user(),
'access_level': 'private',
'retention_policy': '30_days',
'encryption_level': 'standard'
}
if metadata:
default_metadata.update(metadata)
return default_metadata
Context Update and Versioning
```yaml versioning_strategy: version_control: strategy: "semantic_versioning" major_changes: "structural_modifications" minor_changes: "content_additions" patch_changes: "content_updates"
change_tracking: track_changes: true change_log_retention: "90_days" diff_calculation: "intelligent_diff"
conflict_resolution: concurrent_updates: "last_writer_wins_with_merge" conflict_detection: "content_hash_comparison" merge_strategy: "intelligent_merge_with_user_review"
rollback_capability: rollback_support: true rollback_window: "7_days" rollback_granularity: "field_level"
### Context Sharing and Synchronization
#### Intelligent Context Sharing Algorithm
```python
def share_context_between_personas(source_persona, target_persona, context_filter=None):
"""
Intelligently share relevant context between personas
"""
# Get source persona context
source_context = get_persona_context(source_persona)
# Analyze target persona requirements
target_requirements = analyze_persona_context_needs(target_persona)
# Filter relevant context
relevant_context = filter_relevant_context(
source_context,
target_requirements,
context_filter
)
# Apply privacy and security filters
filtered_context = apply_privacy_filters(relevant_context, target_persona)
# Transform context for target persona
transformed_context = transform_context_for_persona(filtered_context, target_persona)
# Validate context compatibility
validate_context_compatibility(transformed_context, target_persona)
# Transfer context
transfer_context(transformed_context, target_persona)
# Log context sharing
log_context_sharing(source_persona, target_persona, transformed_context)
return {
'transfer_successful': True,
'context_items_transferred': len(transformed_context),
'transfer_timestamp': get_current_timestamp()
}
def filter_relevant_context(source_context, target_requirements, context_filter):
"""Filter context based on relevance and requirements"""
relevance_scores = {}
for context_item in source_context:
# Calculate relevance score
relevance_score = calculate_context_relevance(context_item, target_requirements)
# Apply custom filters if provided
if context_filter:
relevance_score = apply_custom_filter(relevance_score, context_item, context_filter)
# Include if above threshold
if relevance_score >= get_relevance_threshold():
relevance_scores[context_item.id] = relevance_score
# Sort by relevance and return top items
sorted_items = sorted(relevance_scores.items(), key=lambda x: x[1], reverse=True)
return [get_context_item(item_id) for item_id, score in sorted_items]
Context Synchronization Strategies
```yaml synchronization_patterns: real_time_sync: description: "Immediate context synchronization" use_cases: ["critical_decisions", "urgent_handoffs", "error_conditions"] latency_target: "< 100ms"
batch_sync: description: "Periodic batch synchronization" use_cases: ["routine_updates", "background_processing", "optimization"] frequency: "every_5_minutes"
event_driven_sync: description: "Synchronization triggered by specific events" use_cases: ["persona_switches", "milestone_completion", "context_changes"] trigger_events: ["persona_handoff", "workflow_transition", "user_action"]
lazy_sync: description: "On-demand synchronization when context is accessed" use_cases: ["infrequent_access", "large_context_sets", "resource_optimization"] cache_strategy: "intelligent_prefetching"
### Context Persistence and Storage
#### Multi-Tier Storage Architecture
\```yaml
storage_tiers:
hot_storage:
description: "Frequently accessed context in memory"
technology: "redis_cluster"
capacity: "active_session_context"
access_time: "< 10ms"
retention: "session_duration"
warm_storage:
description: "Recently accessed context in fast storage"
technology: "ssd_database"
capacity: "recent_project_context"
access_time: "< 50ms"
retention: "30_days"
cold_storage:
description: "Archived context in cost-effective storage"
technology: "object_storage"
capacity: "historical_context"
access_time: "< 500ms"
retention: "1_year"
archive_storage:
description: "Long-term archived context"
technology: "compressed_archive"
capacity: "compliance_retention"
access_time: "< 5_seconds"
retention: "7_years"
Context Persistence Strategies
def persist_context(context_entity, persistence_level='standard'):
"""
Persist context with appropriate storage strategy
"""
# Determine storage tier based on context type and access patterns
storage_tier = determine_storage_tier(context_entity, persistence_level)
# Prepare context for storage
prepared_context = prepare_context_for_storage(context_entity, storage_tier)
# Apply compression if appropriate
if should_compress_context(prepared_context, storage_tier):
prepared_context = compress_context(prepared_context)
# Apply encryption based on sensitivity
if requires_encryption(context_entity):
prepared_context = encrypt_context(prepared_context, context_entity.metadata.encryption_level)
# Store in appropriate tier
storage_result = store_in_tier(prepared_context, storage_tier)
# Update context index
update_context_index(context_entity.id, storage_tier, storage_result.location)
# Set up retention policy
schedule_retention_policy(context_entity.id, context_entity.metadata.retention_policy)
return {
'persistence_successful': True,
'storage_tier': storage_tier,
'storage_location': storage_result.location,
'compression_applied': prepared_context.compressed,
'encryption_applied': prepared_context.encrypted
}
def retrieve_context(context_id, access_pattern='standard'):
"""
Retrieve context with intelligent caching and prefetching
"""
# Check hot cache first
cached_context = check_hot_cache(context_id)
if cached_context:
update_access_statistics(context_id, 'cache_hit')
return cached_context
# Locate context in storage tiers
storage_location = locate_context_in_storage(context_id)
# Retrieve from appropriate tier
stored_context = retrieve_from_storage(storage_location)
# Decrypt if necessary
if stored_context.encrypted:
stored_context = decrypt_context(stored_context)
# Decompress if necessary
if stored_context.compressed:
stored_context = decompress_context(stored_context)
# Cache in hot storage for future access
cache_in_hot_storage(context_id, stored_context)
# Prefetch related context if appropriate
if should_prefetch_related_context(context_id, access_pattern):
prefetch_related_context(context_id)
# Update access statistics
update_access_statistics(context_id, 'storage_retrieval')
return stored_context
Privacy and Security Framework
Access Control and Permissions
```yaml access_control_model: role_based_access: roles: - "context_owner" - "project_member" - "persona_user" - "system_administrator" - "auditor"
permissions:
read: ["view_context_content", "access_context_metadata"]
write: ["modify_context_content", "update_context_metadata"]
share: ["grant_access_to_others", "create_context_links"]
delete: ["remove_context", "purge_context_history"]
admin: ["manage_access_control", "configure_retention_policies"]
attribute_based_access: attributes: - "context_sensitivity_level" - "user_clearance_level" - "project_membership" - "persona_authorization" - "time_based_restrictions"
policies:
- "high_sensitivity_requires_high_clearance"
- "project_context_requires_project_membership"
- "persona_context_requires_persona_authorization"
- "time_restricted_context_enforces_time_limits"
#### Data Privacy and Compliance
```python
def apply_privacy_protection(context_entity, target_persona):
"""
Apply privacy protection based on context sensitivity and target persona
"""
# Assess context sensitivity
sensitivity_level = assess_context_sensitivity(context_entity)
# Get target persona clearance
persona_clearance = get_persona_clearance_level(target_persona)
# Apply privacy filters
if sensitivity_level > persona_clearance:
# Redact sensitive information
filtered_context = redact_sensitive_information(context_entity, sensitivity_level, persona_clearance)
else:
filtered_context = context_entity
# Apply data minimization
minimized_context = apply_data_minimization(filtered_context, target_persona)
# Log privacy protection actions
log_privacy_protection(context_entity.id, target_persona, sensitivity_level, persona_clearance)
return minimized_context
def redact_sensitive_information(context_entity, sensitivity_level, clearance_level):
"""Redact information based on sensitivity and clearance levels"""
redaction_rules = {
'personal_data': ['names', 'emails', 'phone_numbers', 'addresses'],
'financial_data': ['budgets', 'costs', 'revenue_projections'],
'technical_secrets': ['api_keys', 'passwords', 'proprietary_algorithms'],
'business_confidential': ['strategic_plans', 'competitive_analysis', 'internal_processes']
}
redacted_context = copy.deepcopy(context_entity)
for data_type, fields in redaction_rules.items():
if should_redact_data_type(data_type, sensitivity_level, clearance_level):
redacted_context = redact_fields(redacted_context, fields)
return redacted_context
Performance Optimization
Intelligent Caching Strategy
```yaml caching_strategies: multi_level_cache: l1_cache: description: "In-memory cache for immediate access" technology: "application_memory" capacity: "100MB" ttl: "5_minutes"
l2_cache:
description: "Distributed cache for shared access"
technology: "redis_cluster"
capacity: "1GB"
ttl: "30_minutes"
l3_cache:
description: "Persistent cache for warm data"
technology: "ssd_cache"
capacity: "10GB"
ttl: "24_hours"
cache_policies: eviction_policy: "lru_with_priority_boost" prefetch_strategy: "predictive_prefetching" invalidation_strategy: "smart_invalidation"
cache_optimization: compression: "context_aware_compression" serialization: "efficient_binary_serialization" partitioning: "context_type_based_partitioning"
#### Context Retrieval Optimization
```python
def optimize_context_retrieval(context_query, performance_target='standard'):
"""
Optimize context retrieval based on query patterns and performance targets
"""
# Analyze query pattern
query_analysis = analyze_context_query(context_query)
# Determine optimization strategy
optimization_strategy = determine_optimization_strategy(query_analysis, performance_target)
# Apply query optimization
optimized_query = apply_query_optimization(context_query, optimization_strategy)
# Execute with performance monitoring
start_time = get_current_timestamp()
if optimization_strategy.use_parallel_retrieval:
results = execute_parallel_retrieval(optimized_query)
else:
results = execute_sequential_retrieval(optimized_query)
end_time = get_current_timestamp()
retrieval_time = end_time - start_time
# Apply result optimization
optimized_results = optimize_results(results, query_analysis)
# Update performance metrics
update_performance_metrics(context_query, retrieval_time, len(optimized_results))
# Learn from performance
learn_from_retrieval_performance(context_query, optimization_strategy, retrieval_time)
return {
'results': optimized_results,
'retrieval_time': retrieval_time,
'optimization_applied': optimization_strategy,
'performance_target_met': retrieval_time <= get_performance_target(performance_target)
}
Context Intelligence and Learning
Context Pattern Recognition
```yaml pattern_recognition: usage_patterns: frequent_access_patterns: "identify_commonly_accessed_context_combinations" temporal_patterns: "recognize_time_based_context_usage" persona_patterns: "learn_persona_specific_context_preferences" workflow_patterns: "understand_context_flow_in_workflows"
optimization_opportunities: prefetch_candidates: "contexts_likely_to_be_accessed_together" cache_optimization: "contexts_that_benefit_from_longer_caching" compression_candidates: "contexts_suitable_for_compression" archival_candidates: "contexts_ready_for_archival"
anomaly_detection: unusual_access_patterns: "detect_suspicious_context_access" performance_anomalies: "identify_performance_degradation" data_integrity_issues: "detect_context_corruption_or_inconsistency"
#### Adaptive Context Management
```python
def adapt_context_management(usage_statistics, performance_metrics, user_feedback):
"""
Adapt context management strategies based on learning
"""
# Analyze usage patterns
usage_patterns = analyze_usage_patterns(usage_statistics)
# Identify optimization opportunities
optimization_opportunities = identify_optimization_opportunities(
usage_patterns,
performance_metrics
)
# Generate adaptation recommendations
adaptations = generate_adaptation_recommendations(
optimization_opportunities,
user_feedback
)
# Apply safe adaptations automatically
safe_adaptations = filter_safe_adaptations(adaptations)
apply_adaptations(safe_adaptations)
# Queue risky adaptations for review
risky_adaptations = filter_risky_adaptations(adaptations)
queue_for_review(risky_adaptations)
# Monitor adaptation impact
monitor_adaptation_impact(safe_adaptations)
return {
'adaptations_applied': len(safe_adaptations),
'adaptations_queued': len(risky_adaptations),
'expected_performance_improvement': estimate_performance_improvement(adaptations)
}
Monitoring and Analytics
Context Health Monitoring
```yaml health_metrics: availability_metrics: context_availability: "> 99.9%" retrieval_success_rate: "> 99.5%" storage_reliability: "> 99.99%"
performance_metrics: average_retrieval_time: "< 100ms" 95th_percentile_retrieval_time: "< 200ms" cache_hit_ratio: "> 80%"
quality_metrics: context_accuracy: "> 95%" context_completeness: "> 90%" context_relevance: "> 85%"
security_metrics: unauthorized_access_attempts: "< 0.1%" privacy_violations: "0" data_breach_incidents: "0"
#### Analytics and Insights
```python
def generate_context_analytics(time_period='last_30_days'):
"""
Generate comprehensive analytics on context usage and performance
"""
# Collect metrics data
usage_data = collect_usage_metrics(time_period)
performance_data = collect_performance_metrics(time_period)
quality_data = collect_quality_metrics(time_period)
# Generate insights
insights = {
'usage_insights': analyze_usage_trends(usage_data),
'performance_insights': analyze_performance_trends(performance_data),
'quality_insights': analyze_quality_trends(quality_data),
'optimization_recommendations': generate_optimization_recommendations(
usage_data, performance_data, quality_data
)
}
# Create visualizations
visualizations = create_analytics_visualizations(insights)
# Generate reports
reports = generate_analytics_reports(insights, visualizations)
return {
'insights': insights,
'visualizations': visualizations,
'reports': reports,
'generated_at': get_current_timestamp()
}