The Memory Problem: Building Persistent, Queryable Memory for Production AI Agents
By Brian Curry | Vector1 Research
“An agent without memory is not an agent. It is a very expensive stateless function.”
Abstract
The dominant approach to AI agent memory in 2026 is wrong — not wrong in intent, but wrong in architecture. Most production systems treat memory as a retrieval problem: embed interactions, store them in a vector database, retrieve semantically similar chunks at query time. This works well enough for simple factual recall but fails systematically on the problems that matter most in production: temporal reasoning across long horizons, multi-hop relationship traversal, memory consolidation and forgetting, cross-agent memory synchronization, and the management of contradictory or outdated information.
This article presents a complete production architecture for agentic memory that goes beyond retrieval. Drawing on cognitive science’s taxonomy of memory types — episodic, semantic, procedural, and working — and building on the Memory-Node Encapsulation (MNE) data structure introduced in prior Vector1 Research work, we design a four-tier memory system that handles persistence, temporal indexing, consolidation, multi-scope access, and decay. We implement it with a concrete technology stack — Redis, Neo4j, Qdrant, and PostgreSQL — and provide full production code for the components that matter most. The result is a memory architecture that transforms agents from sophisticated stateless functions into systems capable of genuine long-horizon reasoning, relationship awareness, and accumulated expertise.
Keywords: Agentic AI, Memory Architecture, Episodic Memory, Semantic Memory, Memory-Node Encapsulation, MNE, Knowledge Graphs, Vector Retrieval, Multi-Agent Systems, Production AI, LangChain, LangGraph
I. Why Memory Is Still Broken
The state of agentic memory in mid-2026 is best understood by what the benchmarks reveal. The LoCoMo benchmark — 1,540 questions covering single-hop, multi-hop, open-domain, and temporal recall — shows that current systems perform well on single-hop factual recall and poorly on almost everything else. The LongMemEval benchmark, which covers multi-session recall, knowledge updates, and temporal reasoning, reveals a consistent pattern: as conversation history grows, performance degrades faster than context grows. Most systems that score well on accuracy require 26,000+ tokens per query — not production-viable.
Perhaps most damning: a December 2025 benchmark study found that a plain filesystem storing memories as markdown files scored 74% on standard memory tasks — beating dedicated vector-store memory libraries. When a folder of text files outperforms purpose-built memory infrastructure, the infrastructure has a design problem.
The root cause is a category error. Most memory systems are built by ML engineers who think about memory as a retrieval problem. The relevant discipline is actually cognitive architecture — how biological systems store, consolidate, retrieve, and forget information over time. The engineering failure is treating all memory as semantically equivalent and retrieval as the only operation that matters.
Human memory is not a vector database. It has structure that vector similarity cannot capture:
Episodic memory is time-indexed and contextual — not just what happened but when, where, and what else was happening. “The client called angry about the Q3 report three weeks before the contract renewal” is an episodic memory. Its retrieval should be triggered by context (client, contract, report), not just semantic similarity to a query.
Semantic memory is structured, relational, and atemporal — facts and relationships that persist independently of when they were learned. “Client A is in the healthcare vertical, headquartered in Boston, and reports to an IT steering committee” is semantic memory. It lives in a knowledge graph, not a vector index.
Procedural memory is task-knowledge — how to accomplish specific types of work. “When a client escalates a billing issue, always involve the VP of Finance in the first response” is procedural memory. It is triggered by situation type, not content similarity.
Working memory is the active context window — what the agent is currently reasoning about. It is fast, bounded, and temporary.
These memory types require different storage backends, different retrieval strategies, different update mechanisms, and different forgetting curves. Building a production memory system means engineering all four — and the connections between them.
II. The MNE Foundation
Before the full architecture, the core data structure. Memory-Node Encapsulation (MNE), introduced in prior Vector1 Research work, provides the atomic unit on which the entire production memory system is built.
An MNE node encapsulates a memory as a self-contained unit with four components:
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Optional
from enum import Enum
import uuid
import numpy as np
class MemoryType(Enum):
EPISODIC = "episodic" # time-indexed events and interactions
SEMANTIC = "semantic" # facts, relationships, domain knowledge
PROCEDURAL = "procedural" # task patterns and behavioral rules
WORKING = "working" # active context, short-lived
class MemoryStatus(Enum):
ACTIVE = "active"
CONSOLIDATED = "consolidated" # moved from episodic → semantic
DEPRECATED = "deprecated" # superseded by newer information
ARCHIVED = "archived" # retained but low retrieval priority
@dataclass
class MNENode:
"""
Memory-Node Encapsulation: the atomic unit of agentic memory.
Each node represents a single memory — an event, a fact, a procedure,
or a working context item — with full provenance, temporal indexing,
relationship pointers, and a decay model.
The four components:
1. Content: the memory itself (text + structured metadata)
2. Temporal: when it was created, last accessed, and how it decays
3. Relational: connections to other nodes in the memory graph
4. Epistemic: confidence, source quality, and contradiction flags
"""
# Identity
node_id: str = field(default_factory=lambda: str(uuid.uuid4()))
memory_type: MemoryType = MemoryType.EPISODIC
status: MemoryStatus = MemoryStatus.ACTIVE
# Content
content: str = "" # natural language representation
structured: dict[str, Any] = field(default_factory=dict)
embedding: Optional[np.ndarray] = None # dense vector for similarity search
# Temporal
created_at: datetime = field(default_factory=datetime.utcnow)
last_accessed: datetime = field(default_factory=datetime.utcnow)
access_count: int = 0
decay_rate: float = 0.01 # per day — episodic decays faster than semantic
# Scope
agent_id: Optional[str] = None # which agent owns this memory
session_id: Optional[str] = None # which session created it
user_id: Optional[str] = None # which user it's associated with
org_id: Optional[str] = None # organizational scope
# Relational
parent_nodes: list[str] = field(default_factory=list) # generalized from
child_nodes: list[str] = field(default_factory=list) # more specific than
related_nodes: list[str] = field(default_factory=list) # associated with
# Epistemic
confidence: float = 1.0 # 0-1 confidence in this memory
source: str = "" # where this memory came from
contradicts: list[str] = field(default_factory=list) # conflicting node IDs
def salience(self, current_time: datetime) -> float:
"""
Computes current salience: how likely this memory is to be retrieved.
Combines recency, access frequency, and confidence.
Based on the Ebbinghaus forgetting curve, modified for digital systems:
S(t) = confidence * access_boost * exp(-decay_rate * days_since_access)
"""
days_since_access = (
current_time - self.last_accessed
).total_seconds() / 86400
# Access frequency boost (log scale — 10x accesses = 2x boost)
access_boost = 1.0 + 0.5 * np.log1p(self.access_count)
# Ebbinghaus-inspired decay
recency_factor = np.exp(-self.decay_rate * days_since_access)
return float(self.confidence * access_boost * recency_factor)
def access(self) -> None:
"""Record an access — updates recency and count."""
self.last_accessed = datetime.utcnow()
self.access_count += 1The salience() method is the core of the MNE design. Rather than treating all memories as equally retrievable, salience implements a computational forgetting curve — recent, frequently accessed, high-confidence memories are retrieved first. Old, rarely accessed memories fade unless periodically reinforced. This mirrors the neurological basis of human long-term memory retention and solves a real production problem: memory bloat. Without a forgetting mechanism, agent memory grows indefinitely and retrieval quality degrades.
The scope fields (agent_id, session_id, user_id, org_id) enable the multi-scope memory architecture that production systems require — the same memory infrastructure serving multiple agents, users, and organizational contexts simultaneously, with appropriate isolation.
III. The Four-Tier Production Architecture
The complete production architecture runs four storage tiers, each optimized for a specific memory type and access pattern:
┌─────────────────────────────────────────────────────────────────────┐ │ TIER 4: WORKING MEMORY │ │ Redis (in-memory key-value) │ │ Current session context · Active reasoning state · Tool results │ │ TTL: session lifetime (minutes to hours) │ ├─────────────────────────────────────────────────────────────────────┤ │ TIER 3: EPISODIC MEMORY │ │ Qdrant (vector database) + PostgreSQL (temporal index) │ │ Interaction history · Events · Observations │ │ TTL: weeks to months (salience-gated archival) │ ├─────────────────────────────────────────────────────────────────────┤ │ TIER 2: SEMANTIC MEMORY │ │ Neo4j (property graph) │ │ Entities · Relationships · Facts · Domain Knowledge │ │ TTL: indefinite (version-controlled updates) │ ├─────────────────────────────────────────────────────────────────────┤ │ TIER 1: PROCEDURAL MEMORY │ │ PostgreSQL (structured rules) + Qdrant (semantic matching) │ │ Task patterns · Behavioral rules · Learned preferences │ │ TTL: indefinite (reinforcement-updated) │ └─────────────────────────────────────────────────────────────────────┘
3.1 Technology Choices and Rationale
Redis for working memory — sub-millisecond access, TTL-native, supports complex data structures. The agent’s current reasoning state must be available in under 1ms; no other tier can meet this requirement.
Qdrant for episodic and procedural semantic search — the best production vector database for filtered similarity search. The must and should filter clauses allow scope-constrained retrieval (only memories from this user, only from the last 30 days) while maintaining vector similarity ranking.
PostgreSQL for temporal indexing and procedural rules — relational database with native timestamp indexing. Episodic memory is fundamentally a time-series problem; PostgreSQL’s partial indexes and timestamp range queries outperform vector databases for temporal access patterns.
Neo4j for semantic memory — property graphs are the correct data model for entity-relationship knowledge. The question “what do I know about Client A and everyone they’re connected to” is a graph traversal, not a vector similarity search. Neo4j’s Cypher query language handles multi-hop relationship traversal that would require multiple round-trips in a vector database.
3.2 The Memory Manager
The MemoryManager is the single interface through which agents interact with all four tiers:
import redis
import asyncpg
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue
from neo4j import AsyncGraphDatabase
from sentence_transformers import SentenceTransformer
from datetime import datetime, timedelta
from typing import Optional
import json
import numpy as np
class MemoryManager:
"""
Single interface for all four memory tiers.
Handles:
- Write routing: which tier(s) receive a new memory
- Read orchestration: querying multiple tiers and merging results
- Consolidation: promoting episodic → semantic as patterns emerge
- Decay: archiving low-salience memories on a background schedule
- Scope enforcement: isolating memories by agent, user, session, org
Example:
manager = MemoryManager(config)
await manager.remember(node) # write
results = await manager.recall("client issue") # read
await manager.consolidate() # background job
"""
EMBEDDING_DIM = 768
def __init__(self, config: dict):
# Working memory: Redis
self.redis = redis.Redis(
host=config['redis']['host'],
port=config['redis']['port'],
decode_responses=False
)
# Episodic memory: Qdrant + PostgreSQL
self.qdrant = QdrantClient(
host=config['qdrant']['host'],
port=config['qdrant']['port']
)
self._ensure_collections()
# Semantic memory: Neo4j
self.neo4j = AsyncGraphDatabase.driver(
config['neo4j']['uri'],
auth=(config['neo4j']['user'], config['neo4j']['password'])
)
# Embedding model
self.embedder = SentenceTransformer('all-mpnet-base-v2')
# ── Write ──────────────────────────────────────────────────────────
async def remember(self, node: MNENode) -> str:
"""
Write a memory node to the appropriate tier(s).
Routing logic:
- WORKING → Redis only (TTL = session)
- EPISODIC → Qdrant (vector) + PostgreSQL (temporal)
- SEMANTIC → Neo4j (graph)
- PROCEDURAL → PostgreSQL (rules) + Qdrant (semantic match)
"""
# Generate embedding if not already set
if node.embedding is None:
node.embedding = self.embedder.encode(node.content)
if node.memory_type == MemoryType.WORKING:
await self._write_working(node)
elif node.memory_type == MemoryType.EPISODIC:
await self._write_episodic(node)
elif node.memory_type == MemoryType.SEMANTIC:
await self._write_semantic(node)
elif node.memory_type == MemoryType.PROCEDURAL:
await self._write_procedural(node)
return node.node_id
async def _write_episodic(self, node: MNENode) -> None:
"""Write episodic node to Qdrant + PostgreSQL."""
# Qdrant: vector search index
self.qdrant.upsert(
collection_name="episodic",
points=[PointStruct(
id=node.node_id,
vector=node.embedding.tolist(),
payload={
"content": node.content,
"agent_id": node.agent_id,
"user_id": node.user_id,
"session_id": node.session_id,
"created_at": node.created_at.isoformat(),
"confidence": node.confidence,
"memory_type": node.memory_type.value,
"structured": json.dumps(node.structured)
}
)]
)
# PostgreSQL: temporal index for time-range queries
async with self.pg_pool.acquire() as conn:
await conn.execute("""
INSERT INTO episodic_memories
(node_id, agent_id, user_id, session_id, content,
created_at, last_accessed, access_count, confidence,
decay_rate, structured, status)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
ON CONFLICT (node_id) DO UPDATE SET
last_accessed = EXCLUDED.last_accessed,
access_count = episodic_memories.access_count + 1
""",
node.node_id, node.agent_id, node.user_id,
node.session_id, node.content, node.created_at,
node.last_accessed, node.access_count, node.confidence,
node.decay_rate, json.dumps(node.structured),
node.status.value
)
async def _write_semantic(self, node: MNENode) -> None:
"""Write semantic node to Neo4j knowledge graph."""
entity_type = node.structured.get("entity_type", "Concept")
properties = {
"node_id": node.node_id,
"content": node.content,
"confidence": node.confidence,
"source": node.source,
"created_at": node.created_at.isoformat(),
**{k: v for k, v in node.structured.items()
if isinstance(v, (str, int, float, bool))}
}
async with self.neo4j.session() as session:
# Merge entity node
await session.run(f"""
MERGE (n:{entity_type} {{node_id: $node_id}})
SET n += $properties
""", node_id=node.node_id, properties=properties)
# Create relationships to related nodes
for related_id in node.related_nodes:
rel_type = node.structured.get("relationship_type", "RELATED_TO")
await session.run(f"""
MATCH (a {{node_id: $source_id}})
MATCH (b {{node_id: $target_id}})
MERGE (a)-[r:{rel_type}]->(b)
SET r.confidence = $confidence
r.created_at = $created_at
""",
source_id=node.node_id,
target_id=related_id,
confidence=node.confidence,
created_at=node.created_at.isoformat()
)
# ── Read ───────────────────────────────────────────────────────────
async def recall(
self,
query: str,
memory_types: list[MemoryType] | None = None,
agent_id: str | None = None,
user_id: str | None = None,
session_id: str | None = None,
time_range_days: int | None = None,
top_k: int = 10,
min_salience: float = 0.1
) -> list[MNENode]:
"""
Retrieve memories relevant to a query, across all applicable tiers.
The recall pipeline:
1. Embed the query
2. Query each relevant tier with scope filters
3. Merge and deduplicate results
4. Re-rank by composite score: salience × semantic_similarity
5. Return top_k results, update access counts
"""
query_embedding = self.embedder.encode(query)
memory_types = memory_types or [
MemoryType.EPISODIC,
MemoryType.SEMANTIC,
MemoryType.PROCEDURAL
]
results = []
if MemoryType.EPISODIC in memory_types:
episodic = await self._recall_episodic(
query_embedding, agent_id, user_id, session_id,
time_range_days, top_k * 2
)
results.extend(episodic)
if MemoryType.SEMANTIC in memory_types:
semantic = await self._recall_semantic(
query, agent_id, top_k
)
results.extend(semantic)
if MemoryType.PROCEDURAL in memory_types:
procedural = await self._recall_procedural(
query_embedding, agent_id, top_k
)
results.extend(procedural)
# Deduplicate and rank
seen = set()
unique_results = []
for node in results:
if node.node_id not in seen:
seen.add(node.node_id)
unique_results.append(node)
# Composite ranking: salience × cosine similarity
now = datetime.utcnow()
scored = []
for node in unique_results:
if node.embedding is not None:
sim = float(np.dot(query_embedding, node.embedding) /
(np.linalg.norm(query_embedding) *
np.linalg.norm(node.embedding) + 1e-8))
else:
sim = 0.5
sal = node.salience(now)
if sal >= min_salience:
scored.append((node, sal * sim))
scored.sort(key=lambda x: x[1], reverse=True)
top_results = [node for node, _ in scored[:top_k]]
# Update access counts
for node in top_results:
node.access()
await self._update_access(node)
return top_results
async def _recall_episodic(
self,
query_embedding: np.ndarray,
agent_id: str | None,
user_id: str | None,
session_id: str | None,
time_range_days: int | None,
limit: int
) -> list[MNENode]:
"""Retrieve episodic memories from Qdrant with scope + time filters."""
must_conditions = []
if agent_id:
must_conditions.append(
FieldCondition(key="agent_id", match=MatchValue(value=agent_id))
)
if user_id:
must_conditions.append(
FieldCondition(key="user_id", match=MatchValue(value=user_id))
)
# Time range filter (last N days)
if time_range_days:
cutoff = (datetime.utcnow() - timedelta(days=time_range_days)).isoformat()
must_conditions.append(
FieldCondition(key="created_at", range={"gte": cutoff})
)
search_filter = Filter(must=must_conditions) if must_conditions else None
hits = self.qdrant.search(
collection_name="episodic",
query_vector=query_embedding.tolist(),
query_filter=search_filter,
limit=limit,
with_payload=True
)
nodes = []
for hit in hits:
payload = hit.payload
node = MNENode(
node_id=str(hit.id),
memory_type=MemoryType.EPISODIC,
content=payload.get("content", ""),
agent_id=payload.get("agent_id"),
user_id=payload.get("user_id"),
session_id=payload.get("session_id"),
confidence=payload.get("confidence", 1.0),
created_at=datetime.fromisoformat(
payload.get("created_at", datetime.utcnow().isoformat())
),
structured=json.loads(payload.get("structured", "{}")),
)
node.embedding = query_embedding # approximate — recompute if needed
nodes.append(node)
return nodes
async def _recall_semantic(
self,
query: str,
agent_id: str | None,
limit: int
) -> list[MNENode]:
"""Retrieve semantic memories from Neo4j using full-text + relationship traversal."""
async with self.neo4j.session() as session:
result = await session.run("""
CALL db.index.fulltext.queryNodes('memory_content', $query)
YIELD node, score
WHERE ($agent_id IS NULL OR node.agent_id = $agent_id)
AND node.confidence >= 0.5
RETURN node, score
ORDER BY score DESC
LIMIT $limit
""", query=query, agent_id=agent_id, limit=limit)
nodes = []
async for record in result:
props = dict(record["node"])
node = MNENode(
node_id=props.get("node_id", ""),
memory_type=MemoryType.SEMANTIC,
content=props.get("content", ""),
confidence=props.get("confidence", 1.0),
source=props.get("source", ""),
structured={k: v for k, v in props.items()
if k not in ["node_id", "content",
"confidence", "source"]}
)
nodes.append(node)
return nodesIV. Memory Consolidation: Episodic → Semantic
The most important — and most neglected — operation in production memory systems is consolidation: the process by which episodic memories (raw interaction history) are promoted into semantic memories (structured knowledge). Without consolidation, episodic memory grows without bound and semantic memory never reflects what the agent has actually learned.
Biological consolidation happens during sleep. In production agents, it runs as a background job on a configurable schedule.
class MemoryConsolidator:
"""
Promotes episodic memories into semantic knowledge through pattern detection.
The consolidation pipeline:
1. Cluster recent episodic memories by semantic similarity
2. For each cluster above the density threshold, extract the common pattern
3. Check if the pattern contradicts existing semantic memories
4. If novel and consistent: create a new semantic MNE node
5. If contradictory: flag both nodes for human review or update semantic
6. Archive the consolidated episodic nodes (reduce salience, don't delete)
Runs as a scheduled background task — typically nightly or hourly
for high-volume agents.
"""
def __init__(
self,
memory_manager: MemoryManager,
llm_client,
min_cluster_size: int = 3,
similarity_threshold: float = 0.75
):
self.mm = memory_manager
self.llm = llm_client
self.min_cluster_size = min_cluster_size
self.sim_threshold = similarity_threshold
async def consolidate(
self,
agent_id: str,
lookback_days: int = 7
) -> dict:
"""
Run the consolidation pipeline for a specific agent.
Returns statistics on consolidation results.
"""
# 1. Fetch recent episodic memories
recent_episodic = await self._fetch_recent_episodic(
agent_id, lookback_days
)
if len(recent_episodic) < self.min_cluster_size:
return {"status": "skipped", "reason": "insufficient episodic memories"}
# 2. Cluster by semantic similarity
clusters = self._cluster_episodic(recent_episodic)
consolidated_count = 0
contradiction_count = 0
for cluster in clusters:
if len(cluster) < self.min_cluster_size:
continue
# 3. Extract semantic pattern from cluster
pattern = await self._extract_pattern(cluster)
if not pattern:
continue
# 4. Check for contradictions with existing semantic memory
existing = await self.mm.recall(
query=pattern['content'],
memory_types=[MemoryType.SEMANTIC],
agent_id=agent_id,
top_k=3
)
contradiction = self._detect_contradiction(pattern, existing)
if contradiction:
# Flag for review rather than auto-updating
await self._flag_contradiction(pattern, contradiction)
contradiction_count += 1
else:
# Create new semantic memory node
semantic_node = MNENode(
memory_type=MemoryType.SEMANTIC,
content=pattern['content'],
structured=pattern.get('structured', {}),
confidence=pattern.get('confidence', 0.8),
agent_id=agent_id,
source="consolidation",
parent_nodes=[n.node_id for n in cluster],
decay_rate=0.001 # semantic memory decays very slowly
)
await self.mm.remember(semantic_node)
consolidated_count += 1
# Archive source episodic nodes
for episodic_node in cluster:
episodic_node.status = MemoryStatus.CONSOLIDATED
episodic_node.decay_rate *= 3.0 # accelerate decay post-consolidation
await self.mm._update_status(episodic_node)
return {
"status": "completed",
"episodic_processed": len(recent_episodic),
"clusters_found": len(clusters),
"consolidated": consolidated_count,
"contradictions_flagged": contradiction_count
}
async def _extract_pattern(self, cluster: list[MNENode]) -> dict | None:
"""
Use LLM to extract the common semantic pattern from a cluster
of episodic memories.
"""
cluster_text = "\n\n".join([
f"[{i+1}] {node.content}" for i, node in enumerate(cluster[:10])
])
response = await self.llm.complete(f"""
These are {len(cluster)} related interaction memories from an AI agent.
Extract the core semantic pattern or fact they collectively establish.
Memories:
{cluster_text}
Return JSON with:
- "content": a single clear statement of the pattern/fact
- "entity_type": the type of entity this is about (Person, Organization, Process, etc.)
- "confidence": 0-1 confidence that this is a reliable pattern
- "structured": key-value pairs of structured properties
Return null if no clear pattern emerges.
""")
try:
return json.loads(response.text)
except (json.JSONDecodeError, AttributeError):
return None
def _cluster_episodic(
self,
nodes: list[MNENode]
) -> list[list[MNENode]]:
"""Cluster episodic nodes by embedding similarity."""
if not nodes:
return []
embeddings = np.array([
node.embedding for node in nodes
if node.embedding is not None
])
if len(embeddings) < 2:
return [nodes]
from sklearn.cluster import DBSCAN
clustering = DBSCAN(
eps=1 - self.sim_threshold, # cosine distance threshold
min_samples=self.min_cluster_size,
metric='cosine'
).fit(embeddings)
clusters = {}
for i, label in enumerate(clustering.labels_):
if label == -1: # noise point
continue
clusters.setdefault(label, []).append(nodes[i])
return list(clusters.values())V. Multi-Agent Memory Synchronization
Production deployments rarely involve a single agent. A typical enterprise deployment has a coordinator agent, several specialist agents, and potentially user-facing agents — all needing access to a shared memory space with appropriate isolation.
The challenge is memory topology: which agents can read from which memory scopes, how writes propagate, and how conflicts between agents’ memories are resolved.
class MultiAgentMemoryBus:
"""
Coordinates memory access across a multi-agent system.
Memory visibility rules:
- org_id scope: visible to all agents in the organization
- user_id scope: visible to all agents serving this user
- agent_id scope: visible only to the specific agent
- session_id scope: visible only within the current session
Write propagation:
- Agent writes to its own scope
- Consolidation promotes to user or org scope based on generality
- Coordinator agent can explicitly promote memories to wider scope
"""
def __init__(self, memory_manager: MemoryManager):
self.mm = memory_manager
self._subscriptions: dict[str, list[callable]] = {}
async def broadcast_memory(
self,
node: MNENode,
target_scope: str, # 'agent' | 'user' | 'org'
source_agent_id: str
) -> None:
"""
Broadcast a memory to a wider scope.
Used when a specialist agent learns something that all agents should know.
"""
broadcast_node = MNENode(
memory_type=node.memory_type,
content=node.content,
structured=node.structured,
confidence=node.confidence * 0.9, # slight confidence reduction on broadcast
source=f"broadcast_from:{source_agent_id}",
parent_nodes=[node.node_id],
decay_rate=node.decay_rate
)
if target_scope == 'org':
broadcast_node.org_id = node.org_id
broadcast_node.agent_id = None # org-wide
elif target_scope == 'user':
broadcast_node.user_id = node.user_id
broadcast_node.agent_id = None # user-wide
await self.mm.remember(broadcast_node)
# Notify subscribed agents
for callback in self._subscriptions.get(target_scope, []):
await callback(broadcast_node)
def subscribe(self, scope: str, callback: callable) -> None:
"""Subscribe an agent to memory broadcasts at a given scope."""
self._subscriptions.setdefault(scope, []).append(callback)
async def resolve_conflict(
self,
node_a: MNENode,
node_b: MNENode,
resolution_strategy: str = "confidence_weighted"
) -> MNENode:
"""
Resolve conflicting memories from different agents.
Strategies:
- confidence_weighted: weight content by confidence scores
- recency: prefer the more recent memory
- authority: prefer the memory from the designated authoritative agent
"""
if resolution_strategy == "confidence_weighted":
if node_a.confidence >= node_b.confidence:
winner = node_a
loser = node_b
else:
winner = node_b
loser = node_a
# Mark loser as deprecated but don't delete
loser.status = MemoryStatus.DEPRECATED
loser.contradicts.append(winner.node_id)
await self.mm._update_status(loser)
return winner
elif resolution_strategy == "recency":
return node_a if node_a.created_at > node_b.created_at else node_b
return node_a # fallbackVI. The Production Memory Interface: Connecting to LangChain / LangGraph
The architecture above is framework-agnostic. Here is how to connect it to LangChain as a drop-in memory replacement:
from langchain.memory import BaseMemory
from langchain.schema import BaseMessage, HumanMessage, AIMessage
from typing import Dict, List, Any
class MNEMemory(BaseMemory):
"""
LangChain-compatible memory class backed by the full MNE architecture.
Replaces LangChain's built-in memory types with the four-tier
MNE system. Drop-in replacement — existing LangChain chains and
agents work without modification.
Usage:
memory = MNEMemory(
memory_manager=manager,
agent_id="sales_agent_001",
user_id="user_abc",
session_id="session_xyz"
)
chain = LLMChain(llm=llm, prompt=prompt, memory=memory)
"""
memory_manager: Any # MemoryManager instance
agent_id: str
user_id: str
session_id: str
memory_key: str = "chat_history"
return_messages: bool = True
class Config:
arbitrary_types_allowed = True
@property
def memory_variables(self) -> List[str]:
return [self.memory_key]
def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
"""
Load relevant memories for the current input.
Called by LangChain before each LLM invocation.
"""
import asyncio
query = inputs.get("input", inputs.get("question", ""))
# Run async recall in sync context
loop = asyncio.get_event_loop()
memories = loop.run_until_complete(
self.memory_manager.recall(
query=query,
agent_id=self.agent_id,
user_id=self.user_id,
session_id=self.session_id,
top_k=8
)
)
# Format as LangChain messages
messages = []
for mem in memories:
if mem.structured.get("role") == "human":
messages.append(HumanMessage(content=mem.content))
else:
messages.append(AIMessage(content=mem.content))
return {self.memory_key: messages}
def save_context(
self,
inputs: Dict[str, Any],
outputs: Dict[str, str]
) -> None:
"""
Save the current interaction to episodic memory.
Called by LangChain after each LLM invocation.
"""
import asyncio
human_input = inputs.get("input", inputs.get("question", ""))
ai_output = outputs.get("output", outputs.get("response", ""))
loop = asyncio.get_event_loop()
# Save human turn
human_node = MNENode(
memory_type=MemoryType.EPISODIC,
content=human_input,
agent_id=self.agent_id,
user_id=self.user_id,
session_id=self.session_id,
structured={"role": "human"},
decay_rate=0.02
)
loop.run_until_complete(self.memory_manager.remember(human_node))
# Save AI turn
ai_node = MNENode(
memory_type=MemoryType.EPISODIC,
content=ai_output,
agent_id=self.agent_id,
user_id=self.user_id,
session_id=self.session_id,
structured={"role": "ai"},
decay_rate=0.02,
parent_nodes=[human_node.node_id]
)
loop.run_until_complete(self.memory_manager.remember(ai_node))
def clear(self) -> None:
"""Clear working memory for the current session."""
self.memory_manager.redis.delete(f"working:{self.session_id}")VII. Production Considerations
7.1 Memory Drift and Contradiction Management
The most insidious production failure mode is memory drift: the agent’s semantic memory accumulates contradictory or outdated facts without a mechanism to detect or resolve them. An agent that knew a client’s budget was $50K last quarter but has never updated that fact will confidently provide wrong information.
The solution is temporal versioning on all semantic memory nodes. Every update creates a new node version; the previous version is deprecated but retained for audit purposes. Contradiction detection runs as part of the write pipeline, not as a background job:
async def write_with_contradiction_check(
self,
new_node: MNENode,
contradiction_threshold: float = 0.85
) -> tuple[str, list[str]]:
"""
Write a memory node, checking for contradictions first.
Returns the node_id and a list of contradicting node_ids if found.
"""
# Find semantically similar existing memories
similar = await self.recall(
query=new_node.content,
memory_types=[new_node.memory_type],
agent_id=new_node.agent_id,
top_k=5,
min_salience=0.0 # check all, even low-salience
)
contradictions = []
for existing in similar:
if existing.embedding is not None and new_node.embedding is not None:
sim = float(np.dot(new_node.embedding, existing.embedding) /
(np.linalg.norm(new_node.embedding) *
np.linalg.norm(existing.embedding) + 1e-8))
# High similarity but different structured content = likely contradiction
if sim > contradiction_threshold:
if self._content_contradicts(new_node, existing):
contradictions.append(existing.node_id)
existing.contradicts.append(new_node.node_id)
await self._update_status(existing)
new_node.contradicts = contradictions
node_id = await self.remember(new_node)
return node_id, contradictions7.2 Benchmarking Your Memory System
Before shipping a production memory system, validate against the three standard benchmarks:
- LoCoMo (1,540 questions, single-hop, multi-hop, temporal): target >85% accuracy
- LongMemEval (500 questions, knowledge updates, multi-session): target >75% on knowledge update category specifically
- BEAM (1M and 10M token scale): measure latency and token consumption alongside accuracy
The production viability test: if your system requires >5,000 tokens per query at P95, it is not production-viable regardless of accuracy. Optimize for the accuracy/token tradeoff explicitly.
7.3 Deployment Architecture
# docker-compose.yml for local development
services:
redis:
image: redis:7-alpine
ports: ["6379:6379"]
command: redis-server --maxmemory 2gb --maxmemory-policy allkeys-lru
qdrant:
image: qdrant/qdrant:latest
ports: ["6333:6333", "6334:6334"]
volumes: ["./qdrant_storage:/qdrant/storage"]
neo4j:
image: neo4j:5-community
ports: ["7474:7474", "7687:7687"]
environment:
NEO4J_AUTH: neo4j/canary_memory
NEO4J_PLUGINS: '["apoc", "graph-data-science"]'
volumes: ["./neo4j_data:/data"]
postgres:
image: postgres:16-alpine
ports: ["5432:5432"]
environment:
POSTGRES_DB: agent_memory
POSTGRES_USER: agent
POSTGRES_PASSWORD: memory_pass
volumes: ["./pg_data:/var/lib/postgresql/data"]For production on Kubernetes, use managed services: Redis Cloud or Upstash for working memory, Qdrant Cloud for vector search, Neo4j AuraDB for the knowledge graph, and RDS PostgreSQL for temporal indexing.
VIII. Where MNE Goes Next
The architecture in this article represents the production layer of MNE. Several research directions extend it further:
Multimodal memory — extending MNENode to hold image, audio, and structured data embeddings alongside text. MemVerse (December 2025) demonstrated viable multimodal episodic memory; the MNE structure supports this extension natively through the structured dict.
Reinforcement-updated procedural memory — using agent outcome data to update the confidence weights of procedural memory nodes. Procedures that consistently produce good outcomes gain confidence; those associated with failures decay faster.
Cross-agent memory transfer — formalizing the protocol for agents to share memory graphs. The multi-agent bus described in Section V is a starting point; a full transfer protocol would include memory provenance attestation, conflict resolution across organizational boundaries, and privacy-preserving memory sharing.
Memory as fine-tuning signal — using high-salience, high-confidence MNE nodes as training signal for model fine-tuning. The memories an agent returns to most frequently represent the most important learned context; distilling that into model weights is a natural extension.
Getting Started
The MNE data structure and the MemoryManager described in this article are available as part of ongoing Vector1 Research open-source work:
# Coming soon pip install mne-memory
The full architecture is currently in active development. Design discussions, issue tracking, and contribution opportunities are available at github.com/Bodhi8/mne.
For the original MNE specification and cognitive architecture foundations, see the Memory-Node Encapsulation paper on Medium.
References
- Tulving, E. (1972). Episodic and semantic memory. In E. Tulving & W. Donaldson (Eds.), Organization of Memory. Academic Press. [Foundational memory taxonomy]
- Chhikara, P. et al. (2025). Mem0: Building production-ready AI agents with scalable long-term memory. ECAI 2025. arXiv:2504.19413.
- Xu, X. et al. (2025). A-Mem: Agentic memory for LLM agents. arXiv preprint.
- Hu, Y. et al. (2025). A survey on the memory mechanism of large language model based agents. ACM TOIS.
- Kinniment, M. et al. (2024). Evaluating language-model agents on realistic autonomous tasks. NeurIPS 2024.
- Lewis, P. et al. (2020). Retrieval-augmented generation for knowledge-intensive NLP tasks. NeurIPS 2020. [RAG foundations]
- Jiang, D. et al. (2026). MAGMA: A multi-graph based agentic memory architecture. arXiv:2601.03236.
- Huo, Y. et al. (2026). AtomMem: Learnable dynamic agentic memory with atomic memory operation. arXiv:2601.08323.
- Vectorize.io. (2026, March). Best AI agent memory systems in 2026: 8 frameworks compared.
- Mem0.ai. (2026, May). State of AI agent memory 2026: Benchmarks, architectures, and production gaps.
- Ebbinghaus, H. (1885). Über das Gedächtnis [On Memory]. Duncker & Humblot. [Forgetting curve foundations]
- Curry, B. (2024). Memory-Node Encapsulation (MNE): A revolutionary data structure for artificial episodic memory. Vector1 Research / Medium.
- Curry, B. (2024–2026). Vector1 Research Series. Medium / vector1.ai.
About the Author
Brian Curry is a Kansas City–based AI researcher, data scientist, and founder of Vector1 Research, where he works at the intersection of cognitive AI architecture, production agent systems, and knowledge engineering. He is the creator of Memory-Node Encapsulation (MNE), Papilon, and MeaningFlow.
Connect: Medium · LinkedIn · vector1.ai · brian@vector1.ai
