Marcus's Memory System is a sophisticated four-tier learning architecture inspired by cognitive science that transforms every project interaction into accumulated intelligence. It's not just data storage - it's a comprehensive learning engine with Working Memory (immediate awareness), Episodic Memory (specific events), Semantic Memory (general patterns), and Procedural Memory (process optimization) that enables Marcus to learn from every task, predict future challenges, and continuously improve project coordination effectiveness.
Marcus Experience Processing Pipeline
↓
Four-Tier Memory Architecture
↓
┌─────────────────┬─────────────────┬─────────────────┬─────────────────┐
│ Working Memory │ Episodic │ Semantic │ Procedural │
│ (Immediate │ Memory │ Memory │ Memory │
│ Awareness) │ (Events) │ (Patterns) │ (Processes) │
└─────────────────┴─────────────────┴─────────────────┴─────────────────┘
↓ ↓ ↓ ↓
┌─────────────────┬─────────────────┬─────────────────┬─────────────────┐
│ Current Project │ Historical │ Knowledge Base │ Optimization │
│ State & Context │ Event Analysis │ & Pattern Lib │ & Best Practice │
└─────────────────┴─────────────────┴─────────────────┴─────────────────┘
↓
Predictive Intelligence & Continuous Learning
Core Purpose: Transform every project interaction into accumulated intelligence that improves future decision-making, prediction accuracy, and coordination effectiveness.
File: src/core/memory.py - WorkingMemory class
Purpose: Immediate awareness of current project state and active processes
@dataclass
class WorkingMemory:
"""
Immediate awareness layer - what's happening right now
This is Marcus's "short-term memory" of current project state,
active agents, ongoing tasks, and immediate coordination needs.
"""
# Current project state
active_agents: Dict[str, AgentContext] = field(default_factory=dict)
active_tasks: Dict[str, TaskContext] = field(default_factory=dict)
active_blockers: Dict[str, BlockerContext] = field(default_factory=dict)
# Immediate coordination context
pending_assignments: List[AssignmentContext] = field(default_factory=list)
coordination_needs: List[CoordinationContext] = field(default_factory=list)
escalations_in_progress: List[EscalationContext] = field(default_factory=list)
# Real-time intelligence
project_velocity: VelocityMetrics = field(default_factory=VelocityMetrics)
team_capacity: CapacityMetrics = field(default_factory=CapacityMetrics)
risk_indicators: List[RiskIndicator] = field(default_factory=list)Working Memory Operations:
async def update_agent_context(
self,
agent_id: str,
context_update: AgentContextUpdate
) -> None:
"""Update immediate awareness of agent status and context"""
current_context = self.active_agents.get(agent_id, AgentContext())
# Update with new information
current_context.last_activity = context_update.timestamp
current_context.current_tasks = context_update.active_tasks
current_context.availability_status = context_update.status
current_context.recent_performance = context_update.performance_metrics
# Update coordination intelligence
if context_update.indicates_coordination_need():
self.coordination_needs.append(
CoordinationContext(
agent_id=agent_id,
coordination_type=context_update.coordination_type,
urgency=context_update.urgency,
context=context_update.coordination_context
)
)
self.active_agents[agent_id] = current_contextPurpose: Detailed records of specific events, tasks, and coordination instances
@dataclass
class EpisodicMemory:
"""
Event-specific memory - what happened and when
Records detailed information about specific project events,
enabling pattern recognition and experience-based learning.
"""
# Event storage
task_events: List[TaskEvent] = field(default_factory=list)
coordination_events: List[CoordinationEvent] = field(default_factory=list)
blocker_events: List[BlockerEvent] = field(default_factory=list)
agent_interaction_events: List[AgentInteractionEvent] = field(default_factory=list)
async def record_task_completion(
self,
task_completion: TaskCompletionEvent
) -> None:
"""
Record detailed task completion for learning analysis
Captures:
- Task complexity vs actual effort required
- Agent skill match vs performance outcome
- Coordination challenges encountered
- Timeline accuracy vs reality
- Quality indicators and measures
"""
event = TaskEvent(
event_type="task_completion",
task_id=task_completion.task_id,
agent_id=task_completion.agent_id,
completion_context={
"estimated_effort": task_completion.original_estimate,
"actual_effort": task_completion.actual_effort,
"skill_requirements": task_completion.required_skills,
"agent_skills": task_completion.agent_skill_profile,
"coordination_complexity": task_completion.coordination_challenges,
"quality_metrics": task_completion.quality_assessment,
"blockers_encountered": task_completion.blockers,
"learning_opportunities": task_completion.lessons_learned
},
timestamp=task_completion.timestamp,
project_context=task_completion.project_context
)
self.task_events.append(event)
# Trigger pattern analysis for learning
await self._analyze_for_patterns(event)Purpose: General knowledge, patterns, and rules learned from experience
class SemanticMemory:
"""
Pattern and knowledge memory - what we've learned about how things work
Stores general knowledge about project patterns, agent capabilities,
task relationships, and coordination best practices.
"""
def __init__(self):
self.agent_patterns: Dict[str, AgentPattern] = {}
self.task_patterns: Dict[str, TaskPattern] = {}
self.coordination_patterns: Dict[str, CoordinationPattern] = {}
self.blocker_patterns: Dict[str, BlockerPattern] = {}
self.project_patterns: Dict[str, ProjectPattern] = {}
async def update_agent_pattern(
self,
agent_id: str,
performance_data: PerformanceData
) -> None:
"""
Update general knowledge about agent capabilities and patterns
Learns:
- Agent strengths and improvement areas
- Optimal task types for each agent
- Communication and coordination patterns
- Learning velocity and development trends
- Collaboration effectiveness with other agents
"""
pattern = self.agent_patterns.get(agent_id, AgentPattern())
# Update skill assessments
pattern.update_skill_assessment(
skill_data=performance_data.skill_demonstrations,
task_outcomes=performance_data.task_results
)
# Update performance patterns
pattern.update_performance_patterns(
velocity_data=performance_data.velocity_metrics,
quality_data=performance_data.quality_metrics,
consistency_data=performance_data.consistency_metrics
)
# Update coordination patterns
pattern.update_coordination_patterns(
communication_data=performance_data.communication_effectiveness,
collaboration_data=performance_data.collaboration_outcomes
)
self.agent_patterns[agent_id] = pattern
# Generate insights for immediate use
insights = pattern.generate_actionable_insights()
await self._propagate_insights_to_working_memory(insights)
async def learn_task_patterns(
self,
task_type: str,
completion_data: List[TaskCompletionEvent]
) -> None:
"""
Learn general patterns about task types
Discovers:
- Typical effort requirements for task categories
- Common skill combinations needed
- Frequent coordination dependencies
- Risk factors and mitigation strategies
- Quality patterns and best practices
"""
pattern = self.task_patterns.get(task_type, TaskPattern(task_type))
# Analyze effort patterns
effort_analysis = self._analyze_effort_patterns(completion_data)
pattern.effort_expectations.update(effort_analysis)
# Analyze skill patterns
skill_analysis = self._analyze_skill_requirements(completion_data)
pattern.skill_requirements.update(skill_analysis)
# Analyze coordination patterns
coordination_analysis = self._analyze_coordination_needs(completion_data)
pattern.coordination_requirements.update(coordination_analysis)
self.task_patterns[task_type] = patternPurpose: Knowledge about how to do things effectively - process optimization
class ProceduralMemory:
"""
Process and method memory - how to do things effectively
Stores knowledge about effective processes, successful procedures,
and optimization strategies for coordination and project management.
"""
def __init__(self):
self.coordination_procedures: Dict[str, CoordinationProcedure] = {}
self.assignment_procedures: Dict[str, AssignmentProcedure] = {}
self.escalation_procedures: Dict[str, EscalationProcedure] = {}
self.optimization_procedures: Dict[str, OptimizationProcedure] = {}
async def reinforce_successful_procedure(
self,
procedure_type: str,
execution_context: ProcedureExecution,
outcome_assessment: OutcomeAssessment
) -> None:
"""
Reinforce procedures that lead to successful outcomes
Learning process:
1. Identify successful coordination/assignment patterns
2. Extract the procedural elements that contributed to success
3. Reinforce these patterns for future use
4. Generate optimization recommendations
"""
procedure = self.coordination_procedures.get(
procedure_type,
CoordinationProcedure(procedure_type)
)
# Reinforce successful elements
if outcome_assessment.success_score > 0.8:
procedure.reinforce_elements(
successful_elements=execution_context.procedure_elements,
context_factors=execution_context.context,
success_indicators=outcome_assessment.success_factors
)
# Learn optimization opportunities
optimizations = self._identify_optimizations(
execution_context,
outcome_assessment
)
procedure.add_optimizations(optimizations)
# Learn from failures too
elif outcome_assessment.success_score < 0.4:
procedure.learn_from_failure(
failed_elements=execution_context.problematic_elements,
failure_indicators=outcome_assessment.failure_factors,
recovery_actions=execution_context.recovery_attempts
)
self.coordination_procedures[procedure_type] = procedureThe four memory tiers work together to create comprehensive learning:
class MemoryIntegrationEngine:
"""Orchestrates learning across all memory tiers"""
async def process_project_experience(
self,
experience: ProjectExperience
) -> LearningOutcomes:
"""
Process experience through all memory tiers for comprehensive learning
Flow:
1. Working Memory: Immediate context updates
2. Episodic Memory: Detailed event recording
3. Semantic Memory: Pattern extraction and knowledge updates
4. Procedural Memory: Process optimization and best practice refinement
"""
learning_outcomes = LearningOutcomes()
# Working Memory: Update immediate awareness
await self.working_memory.integrate_experience(experience)
learning_outcomes.immediate_insights = self.working_memory.get_immediate_insights()
# Episodic Memory: Record detailed events
episodic_events = await self.episodic_memory.record_experience(experience)
learning_outcomes.event_recordings = episodic_events
# Semantic Memory: Extract and update patterns
pattern_updates = await self.semantic_memory.learn_from_events(episodic_events)
learning_outcomes.pattern_learning = pattern_updates
# Procedural Memory: Optimize processes
procedure_improvements = await self.procedural_memory.optimize_from_patterns(
pattern_updates
)
learning_outcomes.process_optimizations = procedure_improvements
# Cross-tier insights generation
integrated_insights = await self._generate_integrated_insights(
learning_outcomes
)
learning_outcomes.integrated_insights = integrated_insights
return learning_outcomesMemory system enables sophisticated prediction capabilities:
class PredictiveIntelligence:
"""Generate predictions based on accumulated memory"""
async def predict_task_outcome(
self,
agent_id: str,
task: Task
) -> TaskOutcomePrediction:
"""
Predict likely task outcome based on memory patterns
Analysis combines:
- Agent performance patterns from Semantic Memory
- Similar task outcomes from Episodic Memory
- Current context from Working Memory
- Procedural knowledge about assignment effectiveness
"""
# Agent capability assessment
agent_pattern = self.semantic_memory.agent_patterns.get(agent_id)
capability_match = agent_pattern.assess_task_fit(task)
# Historical similar tasks
similar_tasks = self.episodic_memory.find_similar_tasks(
task_type=task.type,
complexity=task.complexity,
skills_required=task.required_skills
)
historical_outcomes = self._analyze_historical_outcomes(similar_tasks)
# Current context factors
current_context = self.working_memory.get_prediction_context(agent_id)
# Procedural effectiveness
assignment_procedure = self.procedural_memory.get_assignment_procedure(
task.type
)
procedure_effectiveness = assignment_procedure.predict_effectiveness(
agent_capability_match=capability_match,
current_context=current_context
)
return TaskOutcomePrediction(
success_probability=self._calculate_success_probability(
capability_match, historical_outcomes, current_context
),
estimated_completion_time=self._predict_completion_time(
agent_pattern, task, similar_tasks
),
risk_factors=self._identify_risk_factors(
agent_pattern, task, current_context
),
optimization_recommendations=procedure_effectiveness.optimizations
)
async def predict_coordination_needs(
self,
project_context: ProjectContext
) -> List[CoordinationPrediction]:
"""
Predict upcoming coordination needs based on memory patterns
Identifies:
- Likely dependency conflicts
- Communication bottlenecks
- Resource contention issues
- Integration challenges
- Timeline coordination needs
"""
predictions = []
# Analyze dependency patterns
dependency_patterns = self.semantic_memory.coordination_patterns.get(
"dependency_coordination"
)
if dependency_patterns:
predicted_conflicts = dependency_patterns.predict_conflicts(
current_dependencies=project_context.active_dependencies,
agent_workloads=project_context.agent_workloads
)
predictions.extend(predicted_conflicts)
# Communication pattern analysis
communication_patterns = self.semantic_memory.coordination_patterns.get(
"communication_coordination"
)
if communication_patterns:
predicted_bottlenecks = communication_patterns.predict_bottlenecks(
team_composition=project_context.team_composition,
project_phase=project_context.current_phase
)
predictions.extend(predicted_bottlenecks)
return predictionsPeriodic consolidation optimizes memory for better learning and prediction:
class MemoryConsolidation:
"""Periodic memory optimization and consolidation"""
async def consolidate_episodic_to_semantic(
self,
time_threshold: timedelta = timedelta(days=30)
) -> ConsolidationResults:
"""
Extract patterns from episodic events and integrate into semantic memory
Process:
1. Identify episodic events older than threshold
2. Extract patterns and trends from event clusters
3. Update semantic memory with discovered patterns
4. Archive or compress old episodic events
"""
old_events = self.episodic_memory.get_events_older_than(time_threshold)
# Pattern extraction
extracted_patterns = {}
for event_cluster in self._cluster_similar_events(old_events):
patterns = self._extract_patterns_from_cluster(event_cluster)
for pattern_type, pattern_data in patterns.items():
if pattern_type not in extracted_patterns:
extracted_patterns[pattern_type] = []
extracted_patterns[pattern_type].append(pattern_data)
# Update semantic memory
consolidation_results = ConsolidationResults()
for pattern_type, pattern_list in extracted_patterns.items():
consolidated_pattern = self._consolidate_patterns(pattern_list)
await self.semantic_memory.integrate_consolidated_pattern(
pattern_type, consolidated_pattern
)
consolidation_results.patterns_consolidated[pattern_type] = len(pattern_list)
# Archive old events
archived_events = await self.episodic_memory.archive_old_events(
old_events, time_threshold
)
consolidation_results.events_archived = len(archived_events)
return consolidation_results
async def optimize_procedural_knowledge(self) -> OptimizationResults:
"""
Optimize procedural knowledge based on recent outcomes
Reviews:
- Procedure effectiveness metrics
- Recent successes and failures
- Optimization opportunities
- Best practice evolution
"""
optimization_results = OptimizationResults()
for procedure_type, procedure in self.procedural_memory.coordination_procedures.items():
# Analyze recent effectiveness
recent_effectiveness = procedure.analyze_recent_effectiveness(
time_window=timedelta(days=14)
)
# Identify optimization opportunities
optimizations = procedure.identify_optimizations(recent_effectiveness)
# Apply optimizations
for optimization in optimizations:
if optimization.confidence_score > 0.8:
procedure.apply_optimization(optimization)
optimization_results.optimizations_applied.append(
f"{procedure_type}: {optimization.description}"
)
# Update procedure
self.procedural_memory.coordination_procedures[procedure_type] = procedure
return optimization_resultsMemory system generates actionable insights for immediate decision-making:
class MemoryInsightEngine:
"""Generate actionable insights from memory analysis"""
async def generate_agent_development_insights(
self,
agent_id: str
) -> List[DevelopmentInsight]:
"""
Generate insights for agent skill development and optimization
Analyzes:
- Agent performance trajectory from Episodic Memory
- Skill development patterns from Semantic Memory
- Current capacity and workload from Working Memory
- Effective development procedures from Procedural Memory
"""
insights = []
# Performance trajectory analysis
agent_pattern = self.semantic_memory.agent_patterns.get(agent_id)
if agent_pattern:
performance_trajectory = agent_pattern.analyze_performance_trajectory()
if performance_trajectory.shows_skill_development_opportunity():
insights.append(DevelopmentInsight(
type="skill_development",
priority="high",
description=f"Agent showing rapid improvement in {performance_trajectory.improving_skills}",
recommendation="Provide more challenging tasks in these areas",
expected_impact="Accelerated skill development and higher performance"
))
# Coordination effectiveness analysis
coordination_patterns = self.semantic_memory.coordination_patterns
agent_coordination = coordination_patterns.get_agent_coordination_patterns(agent_id)
if agent_coordination and agent_coordination.has_improvement_opportunities():
insights.append(DevelopmentInsight(
type="coordination_improvement",
priority="medium",
description=f"Agent could improve in {agent_coordination.improvement_areas}",
recommendation="Pair with high-coordination agents for mentoring",
expected_impact="Better team integration and project flow"
))
return insights
async def generate_project_optimization_insights(
self,
project_context: ProjectContext
) -> List[OptimizationInsight]:
"""
Generate project-level optimization insights from memory analysis
Identifies:
- Resource allocation improvements
- Process optimization opportunities
- Risk mitigation strategies
- Team coordination enhancements
"""
insights = []
# Resource allocation analysis
allocation_patterns = self.semantic_memory.project_patterns.get("resource_allocation")
if allocation_patterns:
current_allocation = project_context.current_resource_allocation
optimal_allocation = allocation_patterns.predict_optimal_allocation(
project_characteristics=project_context.characteristics,
team_composition=project_context.team_composition
)
if allocation_patterns.has_significant_improvement_potential(
current_allocation, optimal_allocation
):
insights.append(OptimizationInsight(
type="resource_reallocation",
priority="high",
description="Current resource allocation suboptimal for project characteristics",
recommendation=f"Reallocate resources according to pattern: {optimal_allocation.summary}",
expected_impact=f"Estimated {optimal_allocation.improvement_percentage}% efficiency increase"
))
return insightsEvery project interaction becomes learning data that improves future performance:
- Task Assignment: Learns optimal agent-task matching patterns
- Progress Tracking: Discovers velocity and quality patterns
- Blocker Resolution: Builds solution libraries and prevention strategies
- Coordination: Develops effective communication and collaboration patterns
Memory enables sophisticated prediction across multiple dimensions:
- Task Outcome Prediction: Success probability, completion time, risk factors
- Agent Performance Prediction: Capability development, workload optimization
- Project Timeline Prediction: Milestone achievement, delay risks, acceleration opportunities
- Coordination Need Prediction: Upcoming bottlenecks, integration challenges
Memory continuously optimizes processes and procedures:
- Assignment Strategies: Learns most effective agent-task matching approaches
- Communication Patterns: Optimizes notification timing and channel selection
- Escalation Procedures: Refines escalation triggers and resolution pathways
- Quality Assurance: Develops quality patterns and improvement strategies
async def enhance_ai_with_memory(
self,
ai_request: AIRequest
) -> MemoryEnhancedAIRequest:
"""
Enhance AI requests with relevant memory context
Provides AI with:
- Historical patterns relevant to current decision
- Similar situations and their outcomes
- Learned preferences and optimization strategies
- Context about agents, tasks, and project patterns
"""
relevant_memories = await self._find_relevant_memories(ai_request)
enhanced_request = ai_request.enhance_with_memory(relevant_memories)
return enhanced_requestasync def inform_assignment_decisions(
self,
assignment_context: AssignmentContext
) -> MemoryGuidedAssignment:
"""
Provide memory-based guidance for task assignments
Includes:
- Agent capability assessments from experience
- Task complexity insights from similar tasks
- Coordination requirements from pattern analysis
- Success probability predictions
"""
memory_guidance = await self._generate_assignment_guidance(assignment_context)
return MemoryGuidedAssignment(assignment_context, memory_guidance)- Repeated mistakes and inefficiencies across projects
- No learning from experience or pattern recognition
- Suboptimal agent-task matching based on limited information
- Reactive coordination without predictive intelligence
- Manual optimization without systematic improvement
- Continuous Learning: Every interaction improves future decision-making
- Predictive Intelligence: Anticipates challenges and optimizes proactively
- Pattern Recognition: Identifies successful strategies and replicates them
- Adaptive Optimization: Continuously improves processes and procedures
- Experience-Based Decisions: All decisions informed by accumulated intelligence
The Memory System transforms Marcus from a reactive task management tool into a continuously learning coordination intelligence. It's the difference between handling each project as a fresh start and building on accumulated experience to deliver increasingly sophisticated and effective project coordination.
Every task assigned, every progress report, every blocker resolved, and every coordination success becomes part of Marcus's growing intelligence, making each subsequent project more efficient, more predictable, and more successful than the last.