neural-memorygraph 1.3.0

Graph-based memory system for AI agents — cognitive science meets deep learning

memorygraph

Graph-based memory system for AI agents — cognitive science meets deep learning.

memorygraph provides a production-ready, research-grade memory layer for LLM agents, autonomous systems, and any AI application that needs to store, retrieve, and reason over structured knowledge across time.

---

Overview

Most agent memory systems are flat key-value stores or simple vector databases. memorygraph treats memory the way cognitive science does — as a dynamic, associative graph where:

• Nodes are memories (semantic facts, episodic events, procedural skills, emotional anchors)
• Edges are typed relationships (causal, hierarchical, associative, temporal, similarity)
• Activation spreads through the graph like priming in the human brain
• Forgetting follows empirically validated curves (Ebbinghaus, Power Law, FSRS-4.5)
• Working memory enforces a bounded attentional buffer with interference effects
• Neural components (Modern Hopfield, DNC, MemN2N, Transformer encoder) are available for learned retrieval

…and it ships ready for the real world:

• One-line agent API — memory.remember(), memory.recall(), memory.reason() with a zero-dependency default embedder
• Framework adapters — LangChain, LangGraph, MCP server, CrewAI, AutoGen, OpenAI & Claude memory backends
• REST API — dependency-free RESTServer plus an optional FastAPI app for production deployments
• Multi-agent memory sharing — one pool, per-agent views with PRIVATE / SHARED / PUBLIC scopes
• Memory timeline — chronological events, time-bucketed histograms, "what did the agent learn when"
• Visualization — NetworkX / Graphviz export, a live web node explorer, and a statistics dashboard (no Flask required)
• Distributed storage — PostgreSQL, Neo4j, Redis and Qdrant backends alongside JSON / Pickle / SQLite
• Benchmark harness — reproducible throughput/latency reports (BENCHMARKS.md)
• 416 tests · 87% coverage — every public surface is unit-tested

Production layer (1.3.0): async/streaming API, REST auth (API-key/JWT/rate-limit/CORS), a fluent metadata query language, graph merge & diff for multi-agent sync, cold-storage archiving, a lifecycle hook/plugin system, real-time WebSocket push, and an optional FAISS ANN index with weight-aware spreading.

---

Architecture

memorygraph/
├── agent.py              # AgentMemory — high-level remember() / recall() / reason() facade
├── aio.py                # AsyncAgentMemory — asyncio-compatible API + streaming
├── embeddings.py         # Pluggable embedders (HashingEmbedder default, OpenAI, SBERT)
├── multiagent.py         # SharedMemoryPool + per-agent views with PRIVATE/SHARED/PUBLIC scopes
├── timeline.py           # MemoryTimeline — chronological events, histograms, span
├── query.py              # MemoryQuery — fluent metadata filter/query language
├── hooks.py              # Lifecycle hook/plugin system (on_remember/recall/forget/...)
├── graphops.py           # merge_graphs / diff_graphs / sync_into for multi-agent sync
├── archive.py            # MemoryArchiver — cold-storage archiving + restore
├── benchmark.py          # Reproducible performance harness → BENCHMARKS.md
│
├── core/
│   ├── types.py          # Enums, protocols, exception hierarchy, type aliases
│   ├── node.py           # MemoryNode — ACT-R activation, retention models, embeddings
│   ├── edge.py           # MemoryEdge — Hebbian weight updates, typed relationships
│   └── graph.py          # MemoryGraph — thread-safe CRUD, search, graph algorithms
│
├── memory/
│   └── working.py        # Baddeley working memory buffer (phonological, visuospatial,
│                         #   episodic, central executive slots)
│
├── retrieval/
│   └── spreading.py      # Collins & Loftus spreading activation (Dijkstra-based)
│
├── decay/
│   └── forgetting.py     # Forgetting models: Ebbinghaus, Power Law, SM-2, FSRS-4.5
│                         # SpacedRepetitionScheduler with per-node review tracking
│
├── neural/               # PyTorch components (optional — graceful degradation)
│   ├── hopfield.py       # Modern Hopfield Networks (Ramsauer et al., 2020)
│   ├── encoder.py        # Transformer memory encoder + contrastive loss
│   └── retriever.py      # DNC-style differentiable memory + End-to-End Memory Networks
│
├── integrations/         # Real-world framework adapters (lazy, optional deps)
│   ├── langchain.py      # BaseMemory + BaseRetriever for LangChain
│   ├── langgraph.py      # BaseStore-compatible semantic store for LangGraph
│   ├── mcp_server.py     # Model Context Protocol server (remember/recall/reason tools)
│   ├── crewai.py         # CrewAI Storage backend
│   ├── autogen.py        # AutoGen Memory protocol (sync + async)
│   ├── openai_memory.py  # OpenAI chat memory injection / capture + RememberingChat
│   └── claude_memory.py  # Anthropic Claude memory backend + memory-tool handler
│
├── viz/                  # Visualization
│   ├── export.py         # to_networkx / to_dot / to_cytoscape / to_d3
│   └── web.py            # Live node explorer + statistics dashboard (stdlib only)
│
├── api/                  # HTTP REST API
│   ├── rest.py           # stdlib RESTServer + optional FastAPI app builder
│   ├── security.py       # API-key/JWT auth, rate limiting, CORS
│   └── websocket.py      # stdlib WebSocket broadcaster (real-time push)
│
└── persistence/
    ├── storage.py        # JSON, Pickle, SQLite backends; backup rotation; npz export
    └── distributed.py    # PostgreSQL, Neo4j, Redis, Qdrant backends

---

Installation

Minimal (NumPy only):

pip install memorygraph

With PyTorch neural components:

pip install "memorygraph[torch]"

Framework integrations (mix and match):

pip install "memorygraph[langchain]"      # or langgraph, mcp, crewai, autogen, openai, anthropic
pip install "memorygraph[integrations]"   # all of the above

Visualization (NetworkX + Graphviz export; the live web explorer needs nothing extra):

pip install "memorygraph[viz]"

Distributed storage backends:

pip install "memorygraph[postgres]"       # or neo4j, redis, qdrant
pip install "memorygraph[distributed]"    # all four

Full installation:

pip install "memorygraph[full]"

Development:

git clone https://github.com/memorygraph/memorygraph
cd memorygraph
pip install -e ".[dev,torch]"

---

Quick Start

Agent Memory in 6 lines

The fastest way in. AgentMemory wraps the whole graph, embedding, association and forgetting machinery behind three verbs — remember, recall, reason — and works out of the box with a deterministic, dependency-free embedder.

from memorygraph import AgentMemory

mem = AgentMemory()                       # zero config; swap in OpenAIEmbedder() for production

mem.remember("The user's name is Leyla", importance=0.95, tags=["user"])
mem.remember("Leyla lives in Istanbul", importance=0.9)
mem.remember("Leyla is a machine-learning engineer")

# Semantic recall
for hit in mem.recall("Where does Leyla live?", k=3):
    print(f"[{hit.score:.2f}] {hit.content}")

# Associative reasoning — spreading activation surfaces *linked* but
# non-matching memories (the foundation of multi-hop recall)
related = mem.reason("Leyla", hops=2)

mem.save("agent_memory.json")             # persist
mem = AgentMemory.load("agent_memory.json")

Plug in a real embedding model for true semantic matching: AgentMemory(embedder=OpenAIEmbedder()) or SentenceTransformerEmbedder(). Runnable end-to-end demos live in examples/.

Basic Memory Graph

import numpy as np
from memorygraph import (
    MemoryGraph, MemoryNode, MemoryEdge,
    MemoryType, EdgeType,
)

# Create a graph with 768-dimensional embeddings (e.g. BERT output size)
graph = MemoryGraph(embedding_dim=768, name="agent_memory")

# Add a semantic memory node
fact = MemoryNode(
    content      = "The Eiffel Tower is located in Paris, France.",
    memory_type  = MemoryType.SEMANTIC,
    embedding    = np.random.randn(768).astype(np.float32),  # use your encoder
    importance   = 0.85,
    tags         = ["geography", "landmarks", "europe"],
)
graph.add_node(fact)

# Add an episodic memory
event = MemoryNode(
    content           = "User asked about European landmarks at 14:32.",
    memory_type       = MemoryType.EPISODIC,
    embedding         = np.random.randn(768).astype(np.float32),
    importance        = 0.6,
    emotional_valence = 0.2,   # slightly positive
)
graph.add_node(event)

# Link them causally
graph.add_edge(MemoryEdge(
    source_id = event.id,
    target_id = fact.id,
    edge_type = EdgeType.CAUSAL,
    weight    = 0.9,
))

print(graph.stats())

Embedding-Based Retrieval

query_embedding = np.random.randn(768).astype(np.float32)

results = graph.similarity_search(
    query     = query_embedding,
    k         = 10,
    threshold = 0.5,
    memory_type = MemoryType.SEMANTIC,  # optional filter
    tags      = ["geography"],          # optional tag filter (AND)
)

for result in results:
    node = graph.get_node(result.node_id)
    print(f"[{result.score:.4f}] {node.content}")

Spreading Activation

Retrieve memories by simulating associative priming — activation flows through the graph from seed nodes, decaying with distance and modulated by edge weights and node importance.

from memorygraph import SpreadingActivation, SpreadingConfig

config = SpreadingConfig(
    max_depth          = 4,
    activation_decay   = 0.6,
    use_edge_weights   = True,
    use_node_importance = True,
    use_temporal_boost = True,   # recently accessed nodes get a boost
    return_top_k       = 20,
    sort_by            = "activation",
)

spreader = SpreadingActivation(graph, config)
activated = spreader.spread(source_ids=[fact.id])

for result in activated:
    node = graph.get_node(result.node_id)
    print(f"[act={result.activation:.4f} depth={result.depth}] {node.content}")

# Cognitive priming: does seeing A make B more accessible?
primed = spreader.priming_search(prime_ids=[event.id], query_id=fact.id)

# Find the conceptual chain between two distant memories
chain = spreader.concept_chain(start_id=node_a.id, end_id=node_b.id)

Working Memory Buffer

A bounded attentional buffer modelled after Baddeley's multi-component working memory model. Supports four slot types with independent capacities, interference between similar items, and LRU/activation-based eviction.

from memorygraph import WorkingMemoryBuffer, WorkingMemorySlot

wm = WorkingMemoryBuffer(
    capacity          = 7,     # Miller's magical number
    decay_rate        = 0.05,
    interference_rate = 0.1,
    eviction_policy   = "lru_activation",
)

# Push nodes into specific slots
wm.push(node,  slot=WorkingMemorySlot.CENTRAL,       activation=1.0, priority=1.0)
wm.push(fact,  slot=WorkingMemorySlot.PHONOLOGICAL,  activation=0.8)
wm.push(event, slot=WorkingMemorySlot.EPISODIC,      activation=0.7)

# Tick to apply decay (call periodically or on each agent step)
wm.tick()

# Inspect the buffer
snapshot = wm.snapshot()
print(f"Utilization: {snapshot['utilization']*100:.0f}%")

# Get the current focus of attention
focus_item = wm.focus()

# Register eviction callback
wm.on_evict = lambda item: print(f"Evicted: {item.node.content}")

Spaced Repetition & Forgetting

Schedule memory reviews using FSRS-4.5, SM-2, Ebbinghaus, or Power Law models. Track stability and difficulty per node.

from memorygraph import SpacedRepetitionScheduler

scheduler = SpacedRepetitionScheduler(model="fsrs", target_retention=0.90)

# Register nodes for spaced repetition
for node in graph.query_by_type(MemoryType.SEMANTIC):
    scheduler.register_node(node.id, initial_grade=node.importance)

# Record a review (grade: 0.0 = forgot, 1.0 = perfect recall)
entry = scheduler.record_review(node_id=fact.id, grade=0.9)
print(f"Next review in {entry.time_until_due() / 86400:.1f} days")
print(f"Stability: {entry.stability:.2f}")

# Get all nodes due for review right now
due_nodes = scheduler.get_due_nodes()

# Get upcoming reviews within the next 3 days
upcoming = scheduler.get_upcoming(within_seconds=3 * 86400)

# Plot the forgetting curve for a specific node
t_array, r_array = scheduler.forgetting_curve_points(
    node_id          = fact.id,
    num_points       = 100,
    time_horizon_days = 30,
)

Persistence

from memorygraph import GraphStorageManager

manager = GraphStorageManager(backend="sqlite")  # or "json", "pickle"
manager.save(graph, "agent_memory.db")

# Load back
graph = manager.load("agent_memory.db")

# Automatic backup rotation (keeps last 3 backups)
manager.save_with_backup("agent_memory.db", keep_backups=3)

# Periodic checkpoints
manager.checkpoint(directory="./checkpoints", prefix="agent")

# Export / import embeddings separately (for fine-tuning pipelines)
manager.export_embeddings(graph, "embeddings.npz")
manager.import_embeddings(graph, "embeddings_v2.npz")

Transactions

with graph.transaction():
    graph.add_node(node_a)
    graph.add_node(node_b)
    graph.add_edge(edge)
    # If any operation raises an exception, the entire transaction rolls back

---

Real-World Integrations

Every adapter imports its framework lazily and falls back to a pure-Python base when the framework is absent — so you can develop and test against them with nothing installed, and they light up automatically once the real package is present.

LangChain

from langchain.chains import ConversationChain
from memorygraph.integrations.langchain import MemoryGraphMemory, MemoryGraphRetriever

# Drop-in conversation memory with semantic recall instead of a blind buffer
memory = MemoryGraphMemory(k=4)
chain = ConversationChain(llm=llm, memory=memory)

# …or use it as a retriever in a RAG chain
retriever = MemoryGraphRetriever(memory.memory, k=5)
docs = retriever.invoke("what are the user's preferences?")

LangGraph

from memorygraph.integrations.langgraph import MemoryGraphStore

store = MemoryGraphStore()                      # BaseStore-compatible, semantically searchable
store.put(("users", "leyla"), "city", {"value": "Istanbul"})
items = store.search(("users",), query="where does she live")

MCP server (Claude Desktop, Cursor, …)

Expose memory to any MCP client as remember / recall / reason / forget / stats tools:

pip install "memorygraph[mcp]"
memorygraph-mcp --store ./agent_memory.json     # stdio MCP server

The tool layer is also transport-agnostic — tool_schema() and dispatch() let you wire the same tools into JSON-RPC / HTTP / WebSocket without the SDK.

CrewAI & AutoGen

from memorygraph.integrations.crewai import MemoryGraphStorage
from crewai.memory import ShortTermMemory
stm = ShortTermMemory(storage=MemoryGraphStorage())

from memorygraph.integrations.autogen import MemoryGraphMemory   # AutoGen Memory protocol
agent = AssistantAgent("assistant", model_client=client, memory=[MemoryGraphMemory()])

OpenAI & Claude memory backends

from openai import OpenAI
from memorygraph.integrations.openai_memory import RememberingChat
chat = RememberingChat(OpenAI(), model="gpt-4o-mini")
chat.send("My name is Leyla and I live in Istanbul.")
chat.send("Where do I live?")        # → recalls "Istanbul" from graph memory

from anthropic import Anthropic
from memorygraph.integrations.claude_memory import RememberingClaude
claude = RememberingClaude(Anthropic(), model="claude-opus-4-8")

ClaudeMemoryAdapter.memory_tool_handler() also implements Anthropic's memory tool contract (create / view / delete), backed by the graph.

---

Visualization

from memorygraph.viz import to_networkx, to_dot, render_graphviz, MemoryGraphServer

G = to_networkx(mem.graph)            # → networkx.DiGraph for centrality, communities, etc.
dot = to_dot(mem.graph)               # → Graphviz DOT string (no dependencies)
render_graphviz(mem.graph, "graph", fmt="png")   # → PNG/SVG/PDF (needs graphviz)

Live node explorer

A built-in, dependency-free web UI (stdlib http.server + D3.js from CDN) with a force-directed graph, semantic search highlighting, and a click-to-inspect detail panel:

from memorygraph.viz import MemoryGraphServer

MemoryGraphServer(mem).serve(port=8080)          # → http://localhost:8080
# or non-blocking:  url = MemoryGraphServer(mem).start_background(port=8080)

Endpoints: /api/graph, /api/search?q=…, /api/node/<id>, /api/stats.

---

Distributed Storage

Beyond the local JSON / Pickle / SQLite backends, four server-backed backends plug into the same GraphStorageManager interface — selected by name or instance:

from memorygraph import GraphStorageManager

# PostgreSQL (JSONB; optional pgvector)
GraphStorageManager(backend="postgres").save(graph, "postgresql://user:pass@host/db")

# Neo4j — nodes/edges map to a real property graph, queryable in Cypher
GraphStorageManager(backend="neo4j").save(graph, "bolt://localhost:7687")

# Redis — fast shared memory with optional TTL-based forgetting
GraphStorageManager(backend="redis").save(graph, "redis://localhost:6379/0")

# Qdrant — scalable ANN vector search over millions of memories
GraphStorageManager(backend="qdrant").save(graph, "http://localhost:6333")

Backend	Best for	path argument
PostgreSQL	durable, transactional, queryable	postgresql://… DSN
Neo4j	native graph queries (Cypher)	bolt://host:7687
Redis	fast, shared, ephemeral (TTL forgetting)	redis://host:6379/0
Qdrant	large-scale semantic / ANN search	http://host:6333

Each backend imports its client lazily and raises a clear error only if used without the driver installed.

---

REST API

Serve memory over HTTP with zero dependencies (stdlib http.server) — or build a full FastAPI app when you want OpenAPI docs and uvicorn scaling.

from memorygraph import AgentMemory
from memorygraph.api import RESTServer

RESTServer(AgentMemory()).serve(port=8000)        # → http://localhost:8000

memorygraph-api --port 8000 --store memory.json   # console entry point

Method	Path	Description
POST	/memories	Create a memory
GET	/memories	List (paginated)
GET	/memories/{id}	One memory + neighbors
DELETE	/memories/{id}	Delete
POST	/recall	Semantic recall {query, k}
POST	/reason	Associative reasoning {cue, hops}
GET	/timeline	Timeline ?bucket=day
GET	/stats	Graph statistics
POST	/forget	Prune low-retention memories

# Production: optional FastAPI app (pip install "memorygraph[api]")
from memorygraph.api import build_fastapi_app
app = build_fastapi_app()          # uvicorn yourmodule:app

---

Multi-Agent Memory Sharing

One shared pool, many agents — each anchored to a room and writing memories at a chosen scope: PRIVATE (owner only), SHARED (same room), or PUBLIC (everyone). Recall and reasoning automatically respect visibility.

from memorygraph import SharedMemoryPool, Scope

pool = SharedMemoryPool()
researcher = pool.agent("researcher", room="growth-team")
analyst    = pool.agent("analyst",    room="growth-team")
intern     = pool.agent("intern",     room="other-team")

researcher.remember("Market grew 30% in 2025", scope=Scope.SHARED)
researcher.remember("Confidential source …",   scope=Scope.PRIVATE)
researcher.remember("Company IPOs in Q3",       scope=Scope.PUBLIC)

analyst.recall("market growth")     # sees SHARED + PUBLIC, not the private note
intern.recall("company")            # other room → only PUBLIC
researcher.share(node_id, Scope.SHARED)   # promote a private memory later

---

Memory Timeline

Inspect when an agent learned or recalled things, straight from each node's temporal metadata.

from memorygraph import MemoryTimeline

tl = MemoryTimeline(mem.graph)
tl.recent(10)                       # last 10 created memories
tl.histogram(bucket="day")          # [{iso, count}, …] for charting
tl.span()                           # (earliest, latest) timestamps
tl.activity_by_type()               # {"episodic": 12, "semantic": 7, …}

The timeline also powers the dashboard and the REST /timeline endpoint.

---

Statistics Dashboard

The visualization server also serves a live statistics dashboard at /dashboard — KPIs plus Chart.js charts for memory-type distribution, importance histogram, creation timeline and degree distribution:

from memorygraph.viz import MemoryGraphServer

srv = MemoryGraphServer(mem)
srv.serve(port=8080)                # /         → node explorer
                                    # /dashboard → statistics dashboard

---

Benchmarks

python benchmarks/run.py 1000 10000      # prints a table and writes BENCHMARKS.md

from memorygraph.benchmark import run_all
report = run_all(sizes=[1000])
print(report.to_markdown())

Measures throughput (ops/s) and latency (ms/op) for remember, recall, reason, similarity_search and persistence. All numbers are single-threaded, CPU-only, using the default HashingEmbedder — a real embedding model with a FAISS index goes much faster.

---

Production Layer (1.3.0)

Async & streaming

from memorygraph import AsyncAgentMemory, gather_recall

mem = AsyncAgentMemory()
await mem.remember("Leyla lives in Istanbul")
hits = await mem.recall("where does Leyla live")          # non-blocking (threadpool)

async for hit in mem.recall_stream("Leyla", k=5):          # streamed results
    print(hit.content)

batches = await gather_recall(mem, ["a", "b", "c"])        # concurrent recalls

The FastAPI app's routes are async too (recall/reason run in a threadpool, never blocking the event loop).

REST security — auth, rate limiting, CORS

from memorygraph.api import RESTServer, SecurityConfig

sec = SecurityConfig.with_api_key(["my-secret"], rate_limit=100, cors_origins=["https://app"])
RESTServer(mem, security=sec).serve(port=8000)
# X-API-Key header (or ?api_key=) required; 401/403 on failure, 429 when rate-limited.

SecurityConfig composes APIKeyAuth, JWTAuth (PyJWT), RateLimiter (sliding window) and CORSConfig. Works on both the stdlib server and the FastAPI app (/health is exempt).

Fluent query / filter API

from memorygraph import MemoryType

results = (
    mem.query()
       .type(MemoryType.EPISODIC)
       .importance(gt=0.7)
       .tag("user")
       .retention(gte=0.3)
       .order_by("importance", descending=True)
       .limit(10)
       .all()
)
hits = mem.query().tag("user").similar_to(vec, k=5)        # hybrid: filter + semantic

Filter on type, importance/retention/valence ranges, tags (all/any), metadata (equals/contains/exists), content substrings, time ranges and consolidation state.

Graph merge & diff (multi-agent sync)

from memorygraph import diff_graphs, merge_graphs, sync_into, MergeStrategy

diff = diff_graphs(agent_a.graph, agent_b.graph)           # added/removed/changed
merged = merge_graphs(agent_a.graph, agent_b.graph, strategy=MergeStrategy.UNION)
sync_into(agent_a.graph, agent_b.graph)                     # in-place sync, returns the diff

Cold-storage archiving

from memorygraph import MemoryArchiver, ArchivePolicy

archiver = MemoryArchiver(mem.graph)
archiver.run(ArchivePolicy(max_importance=0.3, max_retention=0.25))   # offload cold memories
archiver.restore(node_id)                                            # bring one back
archiver.save("cold.json")                                           # persist the archive

Low-importance, low-retention nodes move to a separate archive graph (protecting permanent / high-importance ones), relieving memory pressure on large graphs while staying restorable.

Hooks / plugins

@mem.hooks.on("remember")
def audit(event):
    log.info("new memory: %s", event["node"].content)

from memorygraph.hooks import CounterPlugin
CounterPlugin().attach(mem.hooks)        # telemetry

Events: remember, recall, reason, forget, link, archive. Sync and async handlers supported; handler errors are isolated by default.

Real-time WebSocket push

from memorygraph.api.websocket import WebSocketBroadcaster

ws = WebSocketBroadcaster(port=8765); ws.start()
ws.attach(mem)        # remember/forget/link events now stream to connected clients

const sock = new WebSocket("ws://localhost:8765");
sock.onmessage = e => console.log(JSON.parse(e.data));   // {event:"remember", node:{…}}

Pure stdlib (RFC 6455 handshake + framing) — no websockets dependency.

FAISS index & weight-aware spreading

mem = AgentMemory(use_faiss=True)        # O(log n) ANN search via faiss-cpu (pip install faiss-cpu)

from memorygraph.retrieval.spreading import SpreadingConfig
cfg = SpreadingConfig(normalize_by_outdegree=True,   # ACT-R fan effect: hubs don't dominate
                      aggregate_edges="sum")          # combine parallel edges (max/sum/mean)

---

Testing

pip install -e ".[dev]"
pytest                       # 416 tests, 87% coverage, enforced ≥80% gate

The suite covers the core graph, agent facade, embeddings, every framework adapter (against their dependency-free fallbacks), the REST API (live HTTP round-trip), multi-agent scope isolation, the timeline, the visualization layer and dashboard, the benchmark harness, all forgetting models, working memory, and storage backends.

---

Neural Components (PyTorch)

All neural components require pip install "memorygraph[torch]". They are imported lazily — if PyTorch is not installed, the rest of the library works without modification.

Modern Hopfield Networks

Based on Ramsauer et al. (2020). Exponential storage capacity compared to classical Hopfield networks. Mathematically equivalent to attention with a specific energy function.

from memorygraph.neural import ModernHopfieldLayer, HopfieldMemoryPool, HopfieldStack
import torch

layer = ModernHopfieldLayer(
    input_dim  = 512,
    hidden_dim = 512,
    num_heads  = 8,
    beta       = 4.0,   # inverse temperature (higher = sharper retrieval)
    num_iter   = 3,     # retrieval iterations
)

query    = torch.randn(batch, seq_q, 512)
patterns = torch.randn(batch, seq_k, 512)
output, attn_weights = layer(query, patterns, return_attn=True)

# Energy of the Hopfield network
energy = layer.energy(query, patterns)

# Fixed learnable memory pool
pool = HopfieldMemoryPool(capacity=1024, pattern_dim=512, num_heads=8)
retrieved, weights = pool(query)

# Deep stacked Hopfield with geometric beta schedule
stack = HopfieldStack(
    input_dim     = 512,
    num_layers    = 6,
    beta_schedule = "geometric",
    initial_beta  = 1.0,
    final_beta    = 8.0,
)
output, all_attentions = stack(query, patterns, return_all_attn=True)

Transformer Memory Encoder

from memorygraph.neural import MemoryEncoder, MemoryContrastiveLoss
import torch

encoder = MemoryEncoder(
    vocab_size  = 30522,   # BERT vocabulary
    dim         = 768,
    num_heads   = 12,
    num_layers  = 6,
    max_len     = 512,
    pooling     = "mean",  # or "cls", "max", "attn"
)

input_ids = torch.randint(0, 30522, (batch, seq_len))
output = encoder(input_ids)

print(output.embedding.shape)   # (batch, 768)
print(output.importance)        # (batch, 1)  — learned importance score
print(output.emotion)           # (batch, 1)  — emotional valence
print(output.confidence)        # (batch, 1)  — retrieval confidence

# Contrastive training with InfoNCE / NT-Xent loss
loss_fn = MemoryContrastiveLoss(temperature=0.07)
labels  = torch.arange(batch)
loss    = loss_fn(output.embedding, output.embedding, labels)
loss.backward()

Differentiable Neural Memory (DNC-inspired)

from memorygraph.neural import NeuralMemoryModule
import torch

dnc = NeuralMemoryModule(
    memory_size    = 128,   # number of memory slots
    memory_dim     = 64,    # dimensionality per slot
    controller_dim = 512,   # controller hidden size
    num_reads      = 4,     # parallel read heads
)

# Process a sequence; state carries memory across steps
controller_sequence = torch.randn(batch, time_steps, 512)
output_sequence, final_state = dnc.process_sequence(controller_sequence)
# output_sequence: (batch, time_steps, 512)  controller enriched with memory reads
# final_state.memory: (batch, 128, 64)       memory after all writes

End-to-End Memory Networks (MemN2N)

Based on Sukhbaatar et al. (2015). Multi-hop reasoning over a bag of memory sentences.

from memorygraph.neural import MemoryNetwork
import torch

model = MemoryNetwork(
    vocab_size   = 10000,
    embed_dim    = 256,
    num_hops     = 3,
    output_size  = 10000,   # answer vocabulary
)

query_tokens  = torch.randint(0, 10000, (batch, q_len))
memory_tokens = torch.randint(0, 10000, (batch, num_memories, sent_len))

logits = model(query_tokens, memory_tokens)   # (batch, output_size)

---

Core Concepts

Memory Node Lifecycle

Register → Encode → Store → Retrieve → Reinforce / Decay → Consolidate → (Forget)

Each MemoryNode tracks:

Attribute	Description
importance	0–1 base importance score
emotional_valence	–1 to +1 emotional tone
confidence	0–1 retrieval confidence
consolidation_state	LABILE → CONSOLIDATING → CONSOLIDATED → LONG_TERM
temporal_info	access history, ACT-R activation computation
embedding	dense vector representation
tags	string tags for structured filtering

ACT-R Base-Level Learning:

$$A_i = \ln\left(\sum_j t_j^{-d}\right) + \alpha \cdot e_i \cdot c_i$$

where $t_j$ are times since past accesses, $d$ is the decay parameter, $\alpha$ is emotional boost, $e_i$ is emotional valence, and $c_i$ is confidence.

Retention Models:

Model	Formula	Use Case
Ebbinghaus	$R = e^{-t/S}$	Classic forgetting curve
Power Law	$R = (1 + bt)^{-\alpha}$	Power forgetting (Wixted & Ebbesen)
Hyperbolic	$R = 1/(1 + t/S)$	Hyperbolic discounting
FSRS-4.5	Spaced repetition algorithm	Anki-compatible scheduling

Edge Types

Type	Meaning
ASSOCIATION	General semantic association
CAUSAL	A caused B
TEMPORAL	A preceded B
HIERARCHICAL	A is a kind of / part of B
SIMILARITY	A and B are similar
CONTRADICTS	A contradicts B
COMPOSITIONAL	A is composed of B
EMOTIONAL	A triggered the emotion in B
EPISODIC_LINK	Same episode / context
INFERENCE	B was inferred from A

Consolidation States

LABILE  →  CONSOLIDATING  →  CONSOLIDATED  →  LONG_TERM
  ↑                                                ↓
  └──────────────── RECONSOLIDATING ───────────────┘

Call graph.advance_consolidation(node_id) to progress a node through the consolidation pipeline. Long-term memories are protected from prune_forgotten().

---

Graph Algorithms

# Breadth-first search up to depth 3, only CAUSAL edges
nodes = graph.bfs(start_id, max_depth=3, edge_type=EdgeType.CAUSAL)

# Dijkstra shortest path (cost = 1 - edge_weight)
path = graph.shortest_path(source_id, target_id)

# Strongly connected components (Kosaraju's algorithm)
sccs = graph.strongly_connected_components()

# Merge near-duplicate nodes (cosine similarity > threshold)
merged = graph.merge_similar_nodes(similarity_threshold=0.92)

# Remove forgotten memories below retention threshold
pruned = graph.prune_forgotten(retention_threshold=0.1)

---

Thread Safety

MemoryGraph and WorkingMemoryBuffer are fully thread-safe. All mutating operations acquire a reentrant lock (threading.RLock). The transaction() context manager holds the lock for the duration of the block.

---

Benchmarks

Tested on a consumer laptop (Intel Core i7-12700H, 32 GB RAM) with NumPy embeddings (no GPU):

Operation	N nodes	Time
add_node	10 000	< 1 ms / node
similarity_search (cosine, k=20)	10 000	~12 ms
similarity_search (cosine, k=20)	100 000	~110 ms
spreading_activation (depth=4)	10 000	~8 ms
bfs (depth=3)	10 000	~2 ms
SQLite save	10 000	~1.4 s
SQLite load	10 000	~0.9 s

For sub-millisecond ANN search at scale, install faiss-cpu and swap the embedding index.

---

Design Decisions

Why a graph and not a vector database?
Vector databases excel at "find the 10 most similar things." A graph additionally encodes why memories are related, supports reasoning along paths, models causal and temporal structure, and enables spreading activation — the associative priming that makes retrieval feel natural rather than purely similarity-driven.

Why is PyTorch optional?
The graph core (node CRUD, search, spreading activation, working memory, forgetting models, persistence) has zero heavy dependencies. An agent can run the full cognitive architecture on a Raspberry Pi. PyTorch components are addons for learned retrieval and training pipelines.

Why FSRS-4.5 and not just Ebbinghaus?
FSRS-4.5 is the current state of the art in spaced repetition, used by millions of Anki users. Its stability and difficulty parameters are empirically validated. Ebbinghaus and Power Law are provided for research comparisons and lightweight deployments.

Thread safety via RLock vs. actor model?
An agent's memory is typically accessed by one or two threads (main loop + background consolidation). RLock is simpler, has negligible overhead at this concurrency level, and avoids the latency of message-passing queues.

---

References

• Ramsauer et al. (2020). Hopfield Networks is All You Need. ICLR 2021. arXiv:2008.02217
• Sukhbaatar et al. (2015). End-To-End Memory Networks. NeurIPS 2015. arXiv:1503.08895
• Graves et al. (2016). Hybrid computing using a neural network with dynamic external memory (DNC). Nature 538.
• Anderson et al. (2004). An integrated theory of the mind (ACT-R). Psychological Review.
• Baddeley (2000). The episodic buffer: a new component of working memory. Trends in Cognitive Sciences.
• Collins & Loftus (1975). A spreading-activation theory of semantic processing. Psychological Review.
• Ebbinghaus (1885). Über das Gedächtnis.
• Wixted & Ebbesen (1991). On the form of forgetting. Psychological Science.
• Ye et al. (2022). A New Algorithm for Optimizing Spaced Repetition Scheduling (FSRS). arXiv:2402.10340

---

License

MIT — see LICENSE.