Episodic and Semantic Memory

Introduction

Endel Tulving (1972) classified long-term memory into episodic, semantic, and procedural memory. This classification provides important theoretical guidance for designing agent memory systems. Different types of memory serve different purposes and require different storage and retrieval mechanisms.

Tulving's Memory Classification

Memory Type	Content	Human Example	Agent Example
Episodic	Personal experiences and events	"I went to the park yesterday"	"The user requested data analysis yesterday"
Semantic	General knowledge and facts	"Paris is the capital of France"	"Python's list is a mutable type"
Procedural	Skills and operational procedures	Riding a bicycle	Standard workflow for calling an API

Episodic Memory

Definition and Characteristics

Episodic memory records specific experiences and events, with the following characteristics:

• Temporal markers: Occurred at a specific time
• Context-dependent: Related to a specific scenario
• Autobiographical: Related to "me" (the agent itself)
• Replayable: Can be recalled and re-experienced

Episodic Memory in Agents

Conversation History

The most basic episodic memory -- recording every interaction between the agent and user:

class EpisodicMemory:
    def __init__(self, vector_store):
        self.vector_store = vector_store

    def store_episode(self, episode):
        """Store an interaction event"""
        record = {
            "id": generate_id(),
            "timestamp": datetime.now().isoformat(),
            "user_query": episode["query"],
            "agent_response": episode["response"],
            "tools_used": episode.get("tools", []),
            "outcome": episode.get("outcome", "unknown"),  # success/failure
            "context": episode.get("context", ""),
        }

        # Generate summary for embedding
        summary = f"User asked: {record['user_query']}. Agent responded, outcome: {record['outcome']}"

        self.vector_store.add(
            documents=[summary],
            metadatas=[record],
            ids=[record["id"]]
        )

    def recall(self, query, n=5, time_range=None):
        """Recall relevant experiences"""
        filters = {}
        if time_range:
            filters["timestamp"] = {"$gte": time_range[0], "$lte": time_range[1]}

        return self.vector_store.query(
            query_texts=[query],
            n_results=n,
            where=filters if filters else None
        )

Experience Replay

Borrowing from the reinforcement learning concept, agents can learn from past experiences:

class ExperienceReplay:
    def __init__(self, memory, llm):
        self.memory = memory
        self.llm = llm

    def learn_from_failures(self, current_task):
        """Learn from past failure experiences"""
        # Retrieve similar task failure cases
        failures = self.memory.recall(
            query=current_task,
            n=3,
            filters={"outcome": "failure"}
        )

        if failures:
            lessons = self.llm.generate(
                f"Analyze the following failure cases and extract lessons "
                f"useful for the current task:\n"
                f"Current task: {current_task}\n"
                f"Historical failures: {failures}\n"
                f"Lessons:")
            return lessons
        return None

    def learn_from_successes(self, current_task):
        """Learn from past successful experiences"""
        successes = self.memory.recall(
            query=current_task,
            n=3,
            filters={"outcome": "success"}
        )

        if successes:
            strategy = self.llm.generate(
                f"Based on the following successful cases, recommend a "
                f"strategy for the current task:\n"
                f"Current task: {current_task}\n"
                f"Historical successes: {successes}\n"
                f"Recommended strategy:")
            return strategy
        return None

Memory Stream in Generative Agents

Park et al. (2023) designed the Memory Stream mechanism in "Generative Agents":

class MemoryStream:
    """Generative Agents-style memory stream"""

    def __init__(self):
        self.memories = []

    def add_observation(self, description, importance=None):
        """Add an observation"""
        if importance is None:
            importance = self.score_importance(description)

        self.memories.append({
            "description": description,
            "timestamp": datetime.now(),
            "importance": importance,  # 1-10
            "last_accessed": datetime.now(),
        })

    def retrieve(self, query, n=10):
        """Retrieve based on relevance, recency, and importance"""
        scores = []
        for mem in self.memories:
            recency = self.recency_score(mem)
            importance = mem["importance"] / 10.0
            relevance = self.relevance_score(query, mem["description"])

            # Composite score
            score = recency + importance + relevance
            scores.append((mem, score))

        scores.sort(key=lambda x: x[1], reverse=True)
        return [mem for mem, score in scores[:n]]

    def recency_score(self, memory):
        """Recency score: exponential decay"""
        hours_ago = (datetime.now() - memory["last_accessed"]).total_seconds() / 3600
        return 0.99 ** hours_ago  # Decay factor

Semantic Memory

Definition and Characteristics

Semantic memory stores general knowledge and facts, decoupled from specific experiences:

• Decontextualized: Not dependent on specific times and scenarios
• Structured: Can be organized as concept networks
• Inferrable: Supports logical inference

Knowledge Graphs as Semantic Memory

class SemanticMemory:
    """Knowledge graph-based semantic memory"""

    def __init__(self):
        self.triples = []  # (subject, predicate, object)
        self.entity_index = {}  # entity -> related triples

    def add_knowledge(self, subject, predicate, obj, confidence=1.0):
        """Add a knowledge triple"""
        triple = {
            "subject": subject,
            "predicate": predicate,
            "object": obj,
            "confidence": confidence,
            "source": "observation",
            "updated_at": datetime.now(),
        }
        self.triples.append(triple)

        # Update index
        for entity in [subject, obj]:
            if entity not in self.entity_index:
                self.entity_index[entity] = []
            self.entity_index[entity].append(triple)

    def query(self, subject=None, predicate=None, obj=None):
        """Query knowledge"""
        results = self.triples
        if subject:
            results = [t for t in results if t["subject"] == subject]
        if predicate:
            results = [t for t in results if t["predicate"] == predicate]
        if obj:
            results = [t for t in results if t["object"] == obj]
        return results

    def get_entity_knowledge(self, entity, depth=1):
        """Get knowledge related to an entity (multi-hop capable)"""
        if depth == 0 or entity not in self.entity_index:
            return []

        direct = self.entity_index[entity]
        if depth == 1:
            return direct

        # Multi-hop: get knowledge of neighbor entities
        result = list(direct)
        for triple in direct:
            neighbor = triple["object"] if triple["subject"] == entity else triple["subject"]
            result.extend(self.get_entity_knowledge(neighbor, depth - 1))
        return result

Structured Fact Store

class StructuredFactStore:
    """User-related structured facts"""

    def __init__(self):
        self.facts = {}  # category -> list of facts

    def add_fact(self, category, key, value, source="conversation"):
        if category not in self.facts:
            self.facts[category] = {}

        self.facts[category][key] = {
            "value": value,
            "source": source,
            "confidence": 1.0,
            "updated_at": datetime.now(),
        }

    def get_user_profile(self):
        """Get user profile"""
        return self.facts

    # Usage examples:
    # store.add_fact("preferences", "response_style", "concise")
    # store.add_fact("background", "programming_language", "Python")
    # store.add_fact("project", "current_task", "building RAG system")

Procedural Memory

Definition and Characteristics

Procedural memory stores skills, habits, and operational procedures:

• Implicit knowledge: Usually difficult to describe in words
• Automated: Can be executed automatically once mastered
• Forgetting-resistant: Hard to forget once learned

Agent Procedural Memory

Learned Workflows

class ProceduralMemory:
    """Stores agent's learned operational procedures"""

    def __init__(self, vector_store):
        self.vector_store = vector_store
        self.workflows = {}

    def store_workflow(self, task_type, steps, success_rate=None):
        """Store a successful workflow"""
        workflow = {
            "task_type": task_type,
            "steps": steps,
            "success_rate": success_rate,
            "usage_count": 1,
            "created_at": datetime.now().isoformat(),
        }

        self.workflows[task_type] = workflow

        # Also store in vector DB for semantic retrieval
        description = f"Task type: {task_type}. Steps: {' -> '.join(steps)}"
        self.vector_store.add(
            documents=[description],
            metadatas=[workflow],
            ids=[f"workflow_{task_type}"]
        )

    def find_workflow(self, task_description, threshold=0.8):
        """Find applicable workflow"""
        results = self.vector_store.query(
            query_texts=[task_description],
            n_results=3,
        )

        if results and results["distances"][0][0] < threshold:
            return results["metadatas"][0][0]
        return None

Prompt Template Library

class PromptTemplateMemory:
    """Store and reuse successful prompt templates"""

    def __init__(self):
        self.templates = {}

    def store_template(self, name, template, performance_score):
        self.templates[name] = {
            "template": template,
            "score": performance_score,
            "usage_count": 0,
        }

    def get_best_template(self, task_type):
        """Get the best template for a task type"""
        candidates = {k: v for k, v in self.templates.items() if task_type in k}
        if not candidates:
            return None
        return max(candidates.values(), key=lambda x: x["score"])

Collaboration of Three Memory Types

graph TB
    Q[User Query] --> C{Query Analysis}

    C -->|"Done something similar before?"| EP[Episodic Memory<br/>Retrieve Historical Experience]
    C -->|"What's the relevant knowledge?"| SM[Semantic Memory<br/>Retrieve Knowledge & Facts]
    C -->|"How should this be done?"| PM[Procedural Memory<br/>Retrieve Workflows]

    EP --> INT[Memory Integration]
    SM --> INT
    PM --> INT

    INT --> LLM[LLM Reasoning]
    LLM --> ACT[Action/Answer]

    ACT -->|Record Experience| EP
    ACT -->|Extract Knowledge| SM
    ACT -->|Update Procedures| PM

Practical Application Example

class IntegratedMemory:
    def __init__(self):
        self.episodic = EpisodicMemory(...)
        self.semantic = SemanticMemory()
        self.procedural = ProceduralMemory(...)

    def comprehensive_recall(self, query, task_type=None):
        """Comprehensive recall across all three memory types"""
        context = {}

        # Episodic memory: relevant experiences
        context["past_experiences"] = self.episodic.recall(query, n=3)

        # Semantic memory: relevant knowledge
        entities = extract_entities(query)
        context["knowledge"] = []
        for entity in entities:
            context["knowledge"].extend(
                self.semantic.get_entity_knowledge(entity, depth=2)
            )

        # Procedural memory: applicable workflows
        if task_type:
            context["workflow"] = self.procedural.find_workflow(task_type)

        return context

Memory Updates and Forgetting

Knowledge Conflict Resolution

When new information conflicts with existing memory:

def resolve_conflict(existing_fact, new_fact, llm):
    """Handle knowledge conflicts"""
    if new_fact["source"] == "user_explicit":
        # Explicit user statements have highest priority
        return new_fact

    if new_fact["timestamp"] > existing_fact["timestamp"]:
        # Newer information is typically more reliable
        return new_fact

    # Let LLM judge
    resolution = llm.evaluate(
        f"Existing knowledge: {existing_fact}\nNew information: {new_fact}\n"
        f"Which is more reliable? Why?")
    return resolution

Forgetting Mechanisms

Not all memories need to be permanently retained:

• Temporal decay: Memories not accessed for a long time have lower priority
• Importance filtering: Low-importance memories can be archived or deleted
• Capacity management: When storage approaches limits, clean up least important memories

Further Reading

• Memory Stream and Reflection Mechanisms - Complete memory architecture in Generative Agents
• Tulving, E. (1972). "Episodic and semantic memory"
• Park, J. S., et al. (2023). "Generative Agents: Interactive Simulacra of Human Behavior"
• Shinn, N., et al. (2023). "Reflexion: Language Agents with Verbal Reinforcement Learning"

---

评论