neural-ai-memory 0.3.0

A brain-inspired, dynamic memory system for AI applications

neural-memory

A brain-inspired, dynamic memory layer for AI applications.

Most AI systems treat memory as a static store — embed text, retrieve by similarity. neural-memory goes further: memories have importance scores, decay over time, get reinforced on access, and can be automatically merged or abstracted into higher-level concepts.

from neural_memory import MemorySystem
from neural_memory.providers import LocalProvider

mem = MemorySystem(provider=LocalProvider())

mem.store("The user prefers concise responses.")
mem.store("Project deadline is 2026-05-01", importance=0.9)

results = mem.recall("How should I respond to the user?")
print(results[0].memory.content)
# → "The user prefers concise responses."

---

Installation

No API key required (local embeddings):

pip install neural-ai-memory[local]

With OpenAI:

pip install neural-ai-memory[openai]

With Anthropic (Claude):

pip install neural-ai-memory[anthropic,local]

---

Providers

Provider	Embeddings	LLM scoring	Requires
LocalProvider	sentence-transformers	heuristic	neural-memory[local]
OpenAIProvider	text-embedding-3-small	gpt-4o-mini	neural-memory[openai]
AnthropicProvider	sentence-transformers	Claude Haiku	neural-memory[anthropic,local]

from neural_memory.providers import OpenAIProvider, AnthropicProvider

# OpenAI
mem = MemorySystem(provider=OpenAIProvider(api_key="sk-..."))

# Anthropic
mem = MemorySystem(provider=AnthropicProvider(api_key="sk-ant-..."))

# Custom: subclass BaseLLMProvider and implement embed(), score_importance(), complete()

---

API

store(content, importance=None, context=None, tags=None, metadata=None) → Memory

Ingest and persist a piece of information. Importance is scored automatically unless overridden.

mem.store("Alice is the project lead", tags=["team"])
mem.store("Deploy by Friday", importance=0.95, metadata={"source": "slack"})

store_many(contents, importance=None, ...) → List[Memory]

Batch ingest — embeddings computed in one call instead of N. Significantly faster for large imports.

facts = ["Fact one", "Fact two", "Fact three"]
memories = mem.store_many(facts, importance=0.7)

recall(query, top_k=5, context=None, min_importance=0.0) → List[RetrievalResult]

Retrieve the most relevant memories. Scoring combines semantic relevance, importance, and recency.

results = mem.recall("Who is responsible for deployment?", top_k=3)
for r in results:
    print(f"{r.final_score:.2f}  {r.memory.content}")

forget(memory_id)

Permanently delete a memory.

mem.forget(memory.id)

run_maintenance() → dict

Trigger lifecycle operations manually (also runs automatically every 50 stores):

• Decay — importance drops for memories not accessed recently
• Prune — memories below threshold are deleted
• Merge — highly similar memories are combined into one
• Abstract — clusters of related memories become a single concept

summary = mem.run_maintenance()
# → {"decayed": 12, "pruned": 3, "merged": 2, "abstracted": 1}

stats() → MemoryStats

s = mem.stats()
print(s.total_memories, s.avg_importance, s.by_category)

---

Configuration

from neural_memory import MemorySystem, MemoryConfig, LifecycleConfig, RetrievalWeights

config = MemoryConfig(
    persist_directory="./my_memory",
    retrieval=RetrievalWeights(relevance=0.5, importance=0.3, recency=0.2),
    lifecycle=LifecycleConfig(
        decay_rate=0.05,           # importance lost per day of inactivity
        decay_threshold=0.05,      # memories below this are pruned
        merge_similarity_threshold=0.92,
        abstraction_cluster_size=5,
        auto_maintenance_interval=50,
    ),
    use_llm_importance_scoring=True,
    recency_half_life_days=7.0,
)

mem = MemorySystem(provider=LocalProvider(), config=config)

---

Architecture

store(text)
    │
    ▼
IngestionLayer          normalize, tag detection, metadata
    │
    ▼
EncodingLayer           embed, importance score, category classification
    │
    ├──▶ VectorStorage  ChromaDB — semantic search
    ├──▶ KVStorage      dict/JSON — fast ID lookup
    └──▶ GraphStorage   networkx — relationship tracking

recall(query)
    │
    ▼
RetrievalEngine         weighted score: relevance × 0.5 + importance × 0.3 + recency × 0.2
    │
    ▼
LifecycleManager        reinforce accessed memories

run_maintenance()
    │
    ├── decay + prune
    ├── merge similar pairs
    └── abstract clusters → concept memories

---

Streaming

recall_stream() — sync generator

Yields results one by one, best first. The caller receives the top result immediately without waiting for the full list.

for result in mem.recall_stream("What does the user prefer?", top_k=5):
    print(f"{result.final_score:.2f}  {result.memory.content}")
    # first result prints immediately

arecall_stream() — async generator

async for result in mem.arecall_stream("deployment steps", top_k=3):
    print(result.memory.content)

Provider streaming — complete_stream()

Stream LLM output token by token during merge/abstraction operations. Useful for displaying what the model is doing in real time:

# Custom use: stream a completion from the provider directly
for token in mem._provider.complete_stream("Summarize: A is X. A is also Y."):
    print(token, end="", flush=True)

OpenAIProvider and AnthropicProvider stream real tokens. LocalProvider yields a single fallback chunk.

---

Async API

Every method has an async equivalent — safe to use in FastAPI, asyncio, or any async framework:

from neural_memory import MemorySystem
from neural_memory.providers import LocalProvider

mem = MemorySystem(provider=LocalProvider())

# In an async function:
m = await mem.astore("Deadline is Friday", importance=0.9)
results = await mem.arecall("When is the deadline?")
await mem.aforget(m.id)
summary = await mem.arun_maintenance()

# Batch async:
memories = await mem.astore_many(["Fact A", "Fact B", "Fact C"])

Auto-maintenance triggered by store() always runs in a background thread — it never blocks the calling code.

---

Extending

Implement BaseLLMProvider to add any embedding or LLM backend:

from neural_memory import BaseLLMProvider, MemorySystem

class MyProvider(BaseLLMProvider):
    def embed(self, text: str) -> list[float]:
        ...

    def score_importance(self, content: str, context=None) -> float:
        ...

    def complete(self, prompt: str) -> str:
        ...

mem = MemorySystem(provider=MyProvider())

---

License

MIT