Semantic vs Episodic Memory in AI Systems: Architecture Guide

1. Memory Types: The Foundation

Human memory isn't a single system—it's a sophisticated architecture of specialized subsystems, each optimized for different types of information. AI systems need the same sophistication to match human-level performance.

The three critical memory types for AI agents:

• Semantic Memory: Facts, concepts, and general knowledge ("Python uses indentation for scoping")
• Episodic Memory: Specific events and experiences ("Yesterday's code review identified 3 security issues")
• Procedural Memory: Skills and automated processes ("How to debug a memory leak in Node.js")

Most AI systems today have only episodic memory (conversation logs) with limited semantic memory (RAG documents). The absence of procedural memory—the ability to learn and refine processes—severely limits agent capability.

2. Semantic Memory: Facts & Knowledge

Semantic memory stores facts, concepts, and relationships that are generally true regardless of when or how they were learned. In AI systems, this becomes your agent's knowledge base—the foundational understanding it can apply across contexts.

Characteristics of Semantic Memory

• Context-independent: "FastAPI uses Pydantic for validation" is true regardless of the specific project
• Hierarchical: Concepts build on other concepts (REST > HTTP > Client-Server > Distributed Systems)
• Persistent: Once learned, semantic knowledge should remain stable
• Interconnected: Facts link to other facts through relationships

Implementation with Memory Spine

class SemanticMemoryManager:
    def __init__(self, memory_spine_client):
        self.memory = memory_spine_client
        self.concept_graph = ConceptGraph()
    
    def store_semantic_fact(self, concept, fact, confidence=0.9):
        """Store a semantic fact with concept linkage"""
        
        # Store the fact as a pinned memory (semantic facts should be persistent)
        fact_id = self.memory.pin_critical_memory(
            key=f"semantic_{concept}_{hash(fact)}",
            content=f"Concept: {concept}\nFact: {fact}\nConfidence: {confidence}"
        )
        
        # Add to concept graph for relationship mapping
        self.concept_graph.add_fact(concept, fact, confidence)
        
        # Store cross-references to related concepts
        related_concepts = self._extract_related_concepts(fact)
        for related_concept in related_concepts:
            self.memory.store_memory(
                content=f"Relationship: {concept} relates to {related_concept} via: {fact}",
                tags=["semantic", "relationship", concept, related_concept],
                metadata={
                    "type": "semantic_relationship",
                    "primary_concept": concept,
                    "related_concept": related_concept,
                    "confidence": confidence
                }
            )
    
    def query_semantic_knowledge(self, concept, max_depth=2):
        """Retrieve semantic knowledge about a concept"""
        
        # Get direct facts
        direct_facts = self.memory.search_memories(
            query=f"concept: {concept}",
            limit=20,
            min_confidence=0.7
        )
        
        # Get related concepts through graph traversal
        related_knowledge = {}
        if max_depth > 0:
            related_concepts = self.concept_graph.get_related_concepts(concept, depth=max_depth)
            
            for related_concept in related_concepts:
                related_knowledge[related_concept] = self.memory.search_memories(
                    query=f"concept: {related_concept}",
                    limit=5,
                    min_confidence=0.6
                )
        
        return {
            "direct_facts": direct_facts,
            "related_knowledge": related_knowledge,
            "concept_map": self.concept_graph.get_concept_neighborhood(concept)
        }
    
    def learn_from_interaction(self, interaction_context, extracted_facts):
        """Extract and store semantic knowledge from interactions"""
        
        for fact in extracted_facts:
            # Determine if this is genuinely semantic (general) vs episodic (specific)
            if self._is_semantic_fact(fact):
                concept = self._extract_primary_concept(fact)
                confidence = self._assess_fact_confidence(fact, interaction_context)
                
                # Check if we already know this fact
                existing_knowledge = self.query_semantic_knowledge(concept, max_depth=1)
                
                if self._is_new_or_contradictory(fact, existing_knowledge):
                    self.store_semantic_fact(concept, fact, confidence)
                    
                    # If contradictory, mark for human review
                    if self._is_contradictory(fact, existing_knowledge):
                        self.memory.store_memory(
                            content=f"Potential contradiction: New fact '{fact}' conflicts with existing knowledge",
                            tags=["semantic", "contradiction", "review_needed", concept],
                            metadata={"requires_human_review": True}
                        )

# Example usage
semantic_memory = SemanticMemoryManager(memory_spine_client)

# Store programming concepts
semantic_memory.store_semantic_fact(
    concept="FastAPI",
    fact="FastAPI automatically generates OpenAPI documentation from type hints",
    confidence=0.95
)

semantic_memory.store_semantic_fact(
    concept="Python",
    fact="Python uses garbage collection with reference counting and cycle detection",
    confidence=0.90
)

# Query knowledge
fastapi_knowledge = semantic_memory.query_semantic_knowledge("FastAPI", max_depth=2)
# Returns: direct FastAPI facts + related concepts (Python, REST, Pydantic, etc.)

3. Episodic Memory: Events & Experiences

Episodic memory captures specific events and experiences, complete with temporal and contextual information. For AI agents, this is the history of interactions, decisions made, and outcomes observed.

Implementation with Temporal Context

class EpisodicMemoryManager:
    def __init__(self, memory_spine_client):
        self.memory = memory_spine_client
    
    def store_episode(self, event_type, context, participants, outcome, significance=0.5):
        """Store an episodic memory with rich temporal and contextual metadata"""
        
        episode_content = f"""
        Event Type: {event_type}
        Timestamp: {datetime.utcnow().isoformat()}
        Context: {context}
        Participants: {', '.join(participants)}
        Outcome: {outcome}
        Significance: {significance}
        """
        
        # Store with temporal and participant tags
        episode_id = self.memory.store_memory(
            content=episode_content,
            tags=[
                "episodic",
                event_type,
                f"significance_{int(significance * 10)}"
            ] + [f"participant_{p}" for p in participants],
            metadata={
                "type": "episodic",
                "event_type": event_type,
                "timestamp": datetime.utcnow().isoformat(),
                "participants": participants,
                "significance": significance,
                "context_hash": hashlib.md5(context.encode()).hexdigest()[:8]
            }
        )
        
        return episode_id
    
    def recall_similar_episodes(self, current_context, event_type=None, limit=5):
        """Find similar past episodes for context-aware decision making"""
        
        # Build search query
        search_terms = [current_context]
        if event_type:
            search_terms.append(event_type)
        
        query = " ".join(search_terms)
        
        # Search with episodic filter
        episodes = self.memory.search_memories(
            query=query,
            limit=limit * 2,  # Get more to filter
            min_confidence=0.4
        )
        
        # Filter for episodic memories and rank by relevance
        episodic_memories = [
            ep for ep in episodes 
            if "episodic" in ep.get("tags", [])
        ]
        
        # Sort by combination of similarity and significance
        ranked_episodes = sorted(
            episodic_memories,
            key=lambda ep: (
                ep.get("metadata", {}).get("significance", 0.5) * 
                ep.get("confidence", 0.5)
            ),
            reverse=True
        )
        
        return ranked_episodes[:limit]
    
    def get_temporal_context(self, time_window_hours=24, event_types=None):
        """Get recent episodes for temporal context awareness"""
        
        # Use Memory Spine's timeline feature
        recent_episodes = self.memory.timeline(hours=time_window_hours, limit=50)
        
        # Filter by event types if specified
        if event_types:
            recent_episodes = [
                ep for ep in recent_episodes
                if any(et in ep.get("tags", []) for et in event_types)
            ]
        
        # Group by event type for structured context
        grouped_episodes = {}
        for episode in recent_episodes:
            event_type = episode.get("metadata", {}).get("event_type", "unknown")
            if event_type not in grouped_episodes:
                grouped_episodes[event_type] = []
            grouped_episodes[event_type].append(episode)
        
        return grouped_episodes
    
    def analyze_patterns(self, event_type, lookback_days=30):
        """Analyze patterns in episodic memories to extract insights"""
        
        cutoff_date = datetime.utcnow() - timedelta(days=lookback_days)
        
        # Get all episodes of this type in timeframe
        episodes = self.memory.search_memories(
            query=event_type,
            limit=100,
            min_confidence=0.3
        )
        
        # Filter by date and event type
        filtered_episodes = []
        for episode in episodes:
            episode_date_str = episode.get("metadata", {}).get("timestamp")
            if episode_date_str:
                episode_date = datetime.fromisoformat(episode_date_str.replace('Z', '+00:00'))
                if episode_date >= cutoff_date:
                    filtered_episodes.append(episode)
        
        # Analyze patterns
        patterns = {
            "frequency": len(filtered_episodes) / lookback_days,
            "success_rate": self._calculate_success_rate(filtered_episodes),
            "common_contexts": self._extract_common_contexts(filtered_episodes),
            "avg_significance": self._calculate_avg_significance(filtered_episodes),
            "participant_frequency": self._analyze_participants(filtered_episodes)
        }
        
        return patterns

# Example usage
episodic_memory = EpisodicMemoryManager(memory_spine_client)

# Store a code review episode
episodic_memory.store_episode(
    event_type="code_review",
    context="Pull request #347 - Authentication system refactor",
    participants=["alice", "bob", "ai_agent"],
    outcome="Approved with 2 minor suggestions: add input validation, improve error messages",
    significance=0.7
)

# Later, when doing another code review
similar_reviews = episodic_memory.recall_similar_episodes(
    current_context="Pull request - API authentication changes",
    event_type="code_review",
    limit=3
)

# This helps the agent learn from past review patterns

4. Procedural Memory: Skills & Patterns

Procedural memory is the most sophisticated and arguably most important type for AI agents. It captures how to do things—the processes, workflows, and decision patterns that constitute expertise.

Implementation as Learned Workflows

class ProceduralMemoryManager:
    def __init__(self, memory_spine_client):
        self.memory = memory_spine_client
        self.workflow_patterns = {}
    
    def capture_procedure(self, procedure_name, steps, triggers, success_metrics):
        """Capture a procedure from observed successful executions"""
        
        procedure_content = f"""
        Procedure: {procedure_name}
        
        Triggers: {triggers}
        
        Steps:
        {self._format_steps(steps)}
        
        Success Metrics: {success_metrics}
        
        Captured: {datetime.utcnow().isoformat()}
        """
        
        # Store as pinned procedural memory
        procedure_id = self.memory.pin_critical_memory(
            key=f"procedure_{procedure_name.lower().replace(' ', '_')}",
            content=procedure_content
        )
        
        # Store individual steps for partial matching
        for i, step in enumerate(steps):
            step_content = f"""
            Procedure: {procedure_name} (Step {i+1}/{len(steps)})
            Step: {step['action']}
            Input: {step.get('input', 'N/A')}
            Expected Output: {step.get('expected_output', 'N/A')}
            Conditions: {step.get('conditions', 'N/A')}
            """
            
            self.memory.store_memory(
                content=step_content,
                tags=[
                    "procedural",
                    "step",
                    procedure_name.lower().replace(' ', '_'),
                    f"step_{i+1}"
                ],
                metadata={
                    "type": "procedural_step",
                    "procedure": procedure_name,
                    "step_number": i + 1,
                    "total_steps": len(steps),
                    "action": step['action']
                }
            )
        
        # Cache compiled pattern for fast lookup
        self.workflow_patterns[procedure_name] = {
            "triggers": triggers,
            "steps": steps,
            "success_metrics": success_metrics,
            "usage_count": 0,
            "success_rate": 0.0
        }
        
        return procedure_id
    
    def execute_procedure(self, procedure_name, context, execute_callback):
        """Execute a learned procedure with context adaptation"""
        
        if procedure_name not in self.workflow_patterns:
            # Try to load from memory
            procedure_data = self.memory.get_pinned_memory(
                f"procedure_{procedure_name.lower().replace(' ', '_')}"
            )
            if not procedure_data:
                raise ValueError(f"Unknown procedure: {procedure_name}")
        
        procedure = self.workflow_patterns[procedure_name]
        execution_log = []
        
        try:
            for i, step in enumerate(procedure["steps"]):
                # Adapt step to current context
                adapted_step = self._adapt_step_to_context(step, context)
                
                # Execute through callback
                step_result = execute_callback(adapted_step, context)
                
                execution_log.append({
                    "step": i + 1,
                    "action": adapted_step["action"],
                    "result": step_result,
                    "success": step_result.get("success", False)
                })
                
                # Update context with step results
                context.update(step_result.get("context_updates", {}))
                
                # Check if step failed and handle
                if not step_result.get("success", False):
                    failure_handling = self._handle_step_failure(
                        procedure_name, i, step_result, context
                    )
                    
                    if failure_handling["abort"]:
                        break
                    elif failure_handling["retry"]:
                        step_result = execute_callback(adapted_step, context)
                        execution_log[-1]["result"] = step_result
            
            # Evaluate overall success
            overall_success = all(log["result"].get("success", False) for log in execution_log)
            
            # Update procedure statistics
            self._update_procedure_stats(procedure_name, overall_success)
            
            # Store execution episode for learning
            self._store_execution_episode(procedure_name, context, execution_log, overall_success)
            
            return {
                "success": overall_success,
                "execution_log": execution_log,
                "context": context
            }
            
        except Exception as e:
            self._store_execution_episode(procedure_name, context, execution_log, False, str(e))
            raise
    
    def learn_procedure_variations(self, base_procedure, execution_logs, success_threshold=0.8):
        """Learn procedure variations from successful executions"""
        
        successful_executions = [
            log for log in execution_logs 
            if log["overall_success"] and log["success_rate"] >= success_threshold
        ]
        
        if len(successful_executions) < 3:
            return None  # Need more data
        
        # Analyze variations in successful executions
        variations = self._analyze_execution_variations(successful_executions)
        
        # Create variation patterns
        for variation in variations:
            if variation["frequency"] >= 0.3:  # Appears in 30%+ of successes
                variation_name = f"{base_procedure}_variant_{variation['pattern_id']}"
                
                self.capture_procedure(
                    procedure_name=variation_name,
                    steps=variation["steps"],
                    triggers=variation["triggers"],
                    success_metrics=variation["success_metrics"]
                )
        
        return variations
    
    def get_applicable_procedures(self, context, confidence_threshold=0.7):
        """Find procedures applicable to current context"""
        
        # Search for procedural memories matching context
        relevant_procedures = self.memory.search_memories(
            query=f"context: {context}",
            limit=10,
            min_confidence=confidence_threshold
        )
        
        # Filter for procedural memories
        procedural_matches = []
        for memory in relevant_procedures:
            if "procedural" in memory.get("tags", []):
                procedure_name = memory.get("metadata", {}).get("procedure")
                if procedure_name:
                    # Get full procedure details
                    procedure_details = self.workflow_patterns.get(procedure_name)
                    if procedure_details:
                        procedural_matches.append({
                            "name": procedure_name,
                            "confidence": memory.get("confidence", 0.0),
                            "success_rate": procedure_details.get("success_rate", 0.0),
                            "usage_count": procedure_details.get("usage_count", 0),
                            "triggers": procedure_details.get("triggers", []),
                            "steps": len(procedure_details.get("steps", []))
                        })
        
        # Rank by combination of confidence and success rate
        procedural_matches.sort(
            key=lambda p: p["confidence"] * p["success_rate"] * (1 + p["usage_count"] * 0.1),
            reverse=True
        )
        
        return procedural_matches

# Example: Capturing a debugging procedure
procedural_memory = ProceduralMemoryManager(memory_spine_client)

debugging_procedure = [
    {
        "action": "reproduce_error",
        "input": "error_description",
        "expected_output": "consistent_reproduction",
        "conditions": "same_environment"
    },
    {
        "action": "analyze_logs",
        "input": "log_files",
        "expected_output": "error_patterns",
        "conditions": "sufficient_logging"
    },
    {
        "action": "isolate_cause",
        "input": "error_patterns",
        "expected_output": "root_cause_hypothesis",
        "conditions": "clear_error_pattern"
    },
    {
        "action": "implement_fix",
        "input": "root_cause_hypothesis",
        "expected_output": "code_changes",
        "conditions": "understood_cause"
    },
    {
        "action": "verify_fix",
        "input": "code_changes",
        "expected_output": "error_resolved",
        "conditions": "test_environment_available"
    }
]

procedural_memory.capture_procedure(
    procedure_name="Debug Production Error",
    steps=debugging_procedure,
    triggers=["production_error_reported", "error_rate_spike", "user_complaints"],
    success_metrics=["error_resolved", "no_regression", "fix_time_under_2hours"]
)

5. Memory Spine Implementation

Memory Spine provides specialized tools for implementing all three memory types within a unified system. The key is using the right Memory Spine features for each memory type:

Memory Type Mapping to Memory Spine Features

Unified Memory Architecture

class UnifiedMemorySystem:
    def __init__(self, memory_spine_client):
        self.memory = memory_spine_client
        self.semantic = SemanticMemoryManager(memory_spine_client)
        self.episodic = EpisodicMemoryManager(memory_spine_client)
        self.procedural = ProceduralMemoryManager(memory_spine_client)
    
    def process_interaction(self, user_input, agent_response, context):
        """Process an interaction across all memory types"""
        
        # 1. Extract semantic knowledge (facts that generalize)
        semantic_facts = self._extract_semantic_facts(user_input, agent_response)
        for fact in semantic_facts:
            self.semantic.learn_from_interaction(context, [fact])
        
        # 2. Store episodic memory (this specific interaction)
        self.episodic.store_episode(
            event_type="user_interaction",
            context=f"Topic: {self._extract_topic(user_input)}",
            participants=["user", "agent"],
            outcome=f"Response provided: {agent_response[:100]}...",
            significance=self._assess_interaction_significance(user_input, agent_response)
        )
        
        # 3. Update procedural patterns if applicable
        if self._is_problem_solving_interaction(user_input, agent_response):
            procedure_steps = self._extract_procedure_steps(agent_response)
            if procedure_steps:
                self.procedural.capture_procedure(
                    procedure_name=f"Solve_{self._extract_problem_type(user_input)}",
                    steps=procedure_steps,
                    triggers=[self._extract_problem_type(user_input)],
                    success_metrics=["user_satisfied", "problem_resolved"]
                )
    
    def build_contextual_memory(self, query, max_tokens=4000):
        """Build multi-type memory context for agent use"""
        
        token_budget = {
            "semantic": int(max_tokens * 0.4),    # 40% for facts/knowledge
            "episodic": int(max_tokens * 0.35),   # 35% for relevant experiences
            "procedural": int(max_tokens * 0.25)  # 25% for applicable procedures
        }
        
        context_parts = []
        
        # Semantic context (relevant facts and concepts)
        semantic_knowledge = self.semantic.query_semantic_knowledge(
            concept=self._extract_primary_concept(query),
            max_depth=2
        )
        
        if semantic_knowledge["direct_facts"]:
            semantic_content = self._format_semantic_context(
                semantic_knowledge, 
                token_budget["semantic"]
            )
            context_parts.append(f"Relevant Knowledge:\n{semantic_content}")
        
        # Episodic context (similar past experiences)
        similar_episodes = self.episodic.recall_similar_episodes(
            current_context=query,
            limit=5
        )
        
        if similar_episodes:
            episodic_content = self._format_episodic_context(
                similar_episodes,
                token_budget["episodic"]
            )
            context_parts.append(f"Relevant Experiences:\n{episodic_content}")
        
        # Procedural context (applicable procedures)
        applicable_procedures = self.procedural.get_applicable_procedures(
            context=query,
            confidence_threshold=0.6
        )
        
        if applicable_procedures:
            procedural_content = self._format_procedural_context(
                applicable_procedures,
                token_budget["procedural"]
            )
            context_parts.append(f"Applicable Procedures:\n{procedural_content}")
        
        return "\n\n".join(context_parts)
    
    def memory_consolidation(self):
        """Periodic consolidation across memory types"""
        
        # Consolidate episodic memories (merge similar episodes)
        self.memory.consolidate_memories(decay_threshold=0.3)
        
        # Strengthen semantic memories that appear frequently
        self._strengthen_frequent_semantic_patterns()
        
        # Refine procedural memories based on success rates
        self._refine_procedural_success_patterns()
    
    def get_memory_analytics(self):
        """Get analytics across all memory types"""
        
        return {
            "semantic_facts_count": len(self.semantic.concept_graph.get_all_concepts()),
            "episodic_memories_30d": len(self.episodic.get_temporal_context(24 * 30)),
            "procedural_patterns_count": len(self.procedural.workflow_patterns),
            "cross_type_relationships": self._count_cross_type_relationships(),
            "memory_effectiveness": self._calculate_memory_effectiveness()
        }

# Usage
unified_memory = UnifiedMemorySystem(memory_spine_client)

# Process interactions to build all memory types
unified_memory.process_interaction(
    user_input="How do I optimize PostgreSQL queries?",
    agent_response="Start by analyzing EXPLAIN output, then consider indexes, query structure, and statistics...",
    context={"domain": "database_optimization", "user_level": "intermediate"}
)

# Build context for future interactions
memory_context = unified_memory.build_contextual_memory(
    query="Performance issues with user authentication queries",
    max_tokens=3000
)