Episodic and Semantic Memory for Agents

Memory Taxonomy for Agents

An agent's memory system can be decomposed into four stores, two of which map directly onto the cognitive distinction:

Episodic Memory: Storage and Retrieval

Each experience is stored as a tuple:

Explanation: An episodic trace at time $t$ bundles the observation $s_t$, action $a_t$, reward or outcome $r_t$, any additional metadata $o_t$ (e.g., task ID, user ID), and a timestamp $\tau_t$ — the minimal record of "what happened."

• $s_t$: observation or environment state at step $t$
• $a_t$: action taken by the agent
• $r_t$: scalar reward or structured outcome signal
• $o_t$: auxiliary metadata (session ID, user context, tool results)
• $\tau_t$: wall-clock or logical timestamp

Retrieval is performed by encoding the current query $q$ and scoring stored episodes:

Explanation: The relevance of a stored episode is a weighted combination of semantic similarity to the current query, how recently it occurred, and an importance score (e.g., high-reward events or user-flagged memories) — mirroring how humans preferentially recall vivid or recent events.

• $\phi(\cdot)$: embedding function (e.g., `text-embedding-ada-002` or a local bi-encoder)
• $\cos(\cdot,\cdot)$: cosine similarity in embedding space
• $\text{recency}(\tau_t) = \exp(-\alpha (T - \tau_t))$: exponential decay with rate $\alpha$
• $\text{importance}(e_t)$: scalar assigned at write time (could be $|r_t|$ or an LLM-judged salience score)
• $\lambda_1, \lambda_2, \lambda_3$: mixing weights, tunable per application

Semantic Memory: Distillation from Episodes

Semantic memory is built by periodically consolidating episodic traces into general statements. Formally, given a corpus of episodes $\mathcal{E} = \{e_1, \ldots, e_N\}$, a consolidation function $\mathcal{C}$ produces semantic entries:

Explanation: Semantic entries are generated by clustering related episodes and running a summarization or extraction function $f_\theta$ (typically an LLM call) over each cluster — distilling many specific events into a single general belief, fact, or preference. In practice, $\text{Clusters}(\mathcal{E})$ is computed with k-means or HDBSCAN over episode embeddings $\{\phi(e_i)\}$, and $f_\theta$ is an LLM prompted to extract durable facts (e.g., "User prefers concise answers," "Database schema uses snake_case").

Working Memory Integration

At inference time, the agent assembles its context window by retrieving from both stores:

Explanation: The working context is populated by the top-$k_1$ most relevant episodes, top-$k_2$ most relevant semantic facts, and any matching procedural templates — injecting only the most pertinent long-term memory into the finite context window rather than everything. This selective injection keeps prompt length tractable while giving the LLM backbone access to relevant history and world-knowledge.

PyTorch / LangChain Implementation Sketch

```python import numpy as np from dataclasses import dataclass, field from typing import List, Optional import torch import torch.nn.functional as F @dataclass class Episode: observation: str action: str outcome: str embedding: Optional[torch.Tensor] = None timestamp: float = 0.0 importance: float = 1.0 class EpisodicMemoryStore: def __init__(self, embed_fn, alpha: float = 0.01, lambdas: tuple = (0.6, 0.2, 0.2)): self.episodes: List[Episode] = [] self.embed_fn = embed_fn # callable: str -> torch.Tensor self.alpha = alpha # recency decay rate self.lam = lambdas # (similarity, recency, importance) self.T = 0.0 # logical clock def write(self, obs: str, action: str, outcome: str, importance: float = 1.0): self.T += 1.0 emb = self.embed_fn(f"{obs} {action} {outcome}") ep = Episode(obs, action, outcome, emb, self.T, importance) self.episodes.append(ep) def retrieve(self, query: str, top_k: int = 5) -> List[Episode]: if not self.episodes: return [] q_emb = self.embed_fn(query) # [D] stack = torch.stack([e.embedding for e in self.episodes]) # [N, D]

# Cosine similarity sim = F.cosine_similarity(stack, q_emb.unsqueeze(0)) # [N]

# Recency score timestamps = torch.tensor([e.timestamp for e in self.episodes]) recency = torch.exp(-self.alpha * (self.T - timestamps))

# Importance importance = torch.tensor([e.importance for e in self.episodes]) scores = (self.lam[0] sim + self.lam[1] recency + self.lam[2] * importance / importance.max()) top_idx = scores.topk(min(top_k, len(self.episodes))).indices return [self.episodes[i] for i in top_idx]

class SemanticMemoryStore: """Stores distilled facts as (text, embedding) pairs.""" def __init__(self, embed_fn): self.facts: List[str] = [] self.embeddings: List[torch.Tensor] = [] self.embed_fn = embed_fn def add_fact(self, fact: str): self.facts.append(fact) self.embeddings.append(self.embed_fn(fact)) def retrieve(self, query: str, top_k: int = 3) -> List[str]: if not self.facts: return [] q_emb = self.embed_fn(query) stack = torch.stack(self.embeddings) sim = F.cosine_similarity(stack, q_emb.unsqueeze(0)) top_idx = sim.topk(min(top_k, len(self.facts))).indices return [self.facts[i] for i in top_idx] ```