engramx 0.1.0

EngramX — The open-source agent memory framework with pluggable storage, lifecycle policies, observability, and framework adapters.

The Open-Source Agent Memory Framework

---

Why EngramX?

An engram (from the Greek engramma, "that which is written on") is the neuroscience term for the physical trace a memory leaves in the brain. EngramX brings the same idea to AI agents: structured, durable, queryable memory that works across any framework or backend.

The current ecosystem for agent memory is fragmented, framework-locked, and under-tooled. Every platform defines memory differently, uses incompatible schemas, and provides minimal support for debugging, governance, or cross-framework composability. EngramX is the neutral abstraction layer that sits under or beside any of them.

How EngramX Compares

Dimension	Mem0	Zep	Letta/MemGPT	LangChain	EngramX
Common memory schema	Proprietary	Proprietary	Proprietary	Per-pattern	Canonical
Episodic / Semantic / Procedural	Partial	Partial	Partial (tiers)	Partial	First-class
Pluggable backends	Limited	No	No	Partial	9 drivers
Declarative lifecycle policies	Partial	No	No	No	YAML DSL
Retrieval explainability	Dashboard	No	ADE only	No	Built-in API
Right-to-be-forgotten	Cloud tier	No	No	No	Policy-driven
Framework adapters	SDK only	API only	No	Native only	LangChain, LlamaIndex, AutoGen
Built-in test harness	No	No	No	No	MemoryHarness
Composable with others	No	No	No	No	Wraps Mem0/Zep

---

Features

• Canonical schema — MemoryRecord with episodic, semantic, procedural, and meta types
• 9 storage backends — SQLite, PostgreSQL + pgvector, ChromaDB, Qdrant, Redis, Neo4j, Mem0, Zep, and in-memory
• Policy engine — Declarative YAML rules for extraction, retention decay, summarization/promotion, and GDPR governance
• Observability — Retrieval traces, memory timelines, output attribution, and explain() as a first-class API
• Hybrid retrieval — Lexical + vector + importance + decay scoring with configurable modes
• Embedding providers — OpenAI, Gemini, Cohere, Mistral, Together AI, LiteLLM (100+ providers), Sentence Transformers, and built-in hash embedder
• LLM reflection — Pluggable reflectors (OpenAI, Anthropic, LiteLLM, or any OpenAI-compatible provider via base_url) for summarizing episodic memory into durable knowledge
• Framework adapters — Thin integrations for LangChain, LlamaIndex, and AutoGen
• Agent runner — EngramAgent with run_with_attribution() for tracking which memories influenced output
• Background jobs — PolicyJobScheduler for scheduled decay, promotion, and governance enforcement
• Test harness — MemoryHarness for testing memory behavior in CI
• Async-first — Full async API with sync wrappers for convenience

---

Installation

# pip
pip install engramx

# uv
uv add engramx

Install latest alpha directly from GitHub:

# pip
pip install git+https://github.com/TechyNilesh/EngramX.git

# uv
uv add git+https://github.com/TechyNilesh/EngramX.git

Install with optional backends:

pip install engramx[postgres]           # PostgreSQL + pgvector
pip install engramx[chroma]             # ChromaDB
pip install engramx[qdrant]             # Qdrant
pip install engramx[redis]              # Redis
pip install engramx[neo4j]              # Neo4j
pip install engramx[mem0]               # Mem0 delegation
pip install engramx[zep]                # Zep delegation
pip install engramx[openai]             # OpenAI embeddings + reflection
pip install engramx[anthropic]          # Anthropic reflection
pip install engramx[litellm]            # LiteLLM (100+ providers)
pip install engramx[sentence-transformers]  # Local embeddings
pip install engramx[langchain]          # LangChain adapter
pip install engramx[llamaindex]         # LlamaIndex adapter
pip install engramx[autogen]            # AutoGen adapter
pip install engramx[langgraph]          # LangGraph agent memory
pip install engramx[all]                # Everything

Install from source (development):

git clone https://github.com/TechyNilesh/EngramX.git
cd EngramX
pip install -e .[dev]

---

Quick Start

import asyncio
from engramx import MemoryClient

async def main() -> None:
    client = MemoryClient(driver="memory")

    # Store a memory
    memory_id = await client.add(
        type="semantic",
        scope="user",
        user_id="u-123",
        content="User prefers metric units for measurements.",
        source="conversation",
        importance_score=0.8,
        sensitivity_level="internal",
    )

    # Search memories
    results = await client.search(
        query="unit preferences",
        filters={"scope": "user", "user_id": "u-123", "type": "semantic"},
        top_k=5,
    )

    # Explain retrieval
    trace = await client.explain(
        query="unit preferences",
        filters={"user_id": "u-123"},
    )

    print(results[0].record.content)
    print(trace.to_markdown())

asyncio.run(main())

---

Storage Backends

EngramX decouples the API from storage via a BaseDriver interface. All 9 drivers implement the same contract.

Driver	Backend	Best for	Install extra
InMemoryDriver	Python dict	Tests, prototyping	—
SQLiteDriver	SQLite	Local dev, single-agent	—
PostgresDriver	PostgreSQL + pgvector	Production, vector search	postgres
ChromaDriver	ChromaDB	Lightweight semantic search	chroma
QdrantDriver	Qdrant	High-performance vector search	qdrant
RedisDriver	Redis / Valkey	Fast short-term memory, caching	redis
Neo4jDriver	Neo4j	Graph-structured memory, temporal edges	neo4j
Mem0Driver	Mem0 API	Delegate to Mem0 + EngramX observability	mem0
ZepDriver	Zep API	Delegate to Zep's temporal knowledge graph	zep

from engramx import MemoryClient

# Use any driver by name
client = MemoryClient(driver="sqlite")
client = MemoryClient(driver="postgres")
client = MemoryClient(driver="chroma")
client = MemoryClient(driver="qdrant")
client = MemoryClient(driver="redis")
client = MemoryClient(driver="neo4j")

# Or pass a driver instance directly
from engramx import PostgresDriver
client = MemoryClient(PostgresDriver(dsn="postgresql://localhost/engramx"))

---

Memory Types

EngramX enforces meaningful separation of three cognitive memory types:

Episodic — Raw experiences: conversation turns, tool calls, agent decisions.

await client.add(type="episodic", scope="session", content="Payment API returned 402.", source="tool_call")

Semantic — Distilled facts: user preferences, domain knowledge, stable truths.

await client.add(type="semantic", scope="user", content="User prefers metric units.", source="reflection")

Procedural — Learned skills: action templates and decision heuristics from repeated success.

await client.add(type="procedural", scope="agent", content="Check visa before searching flights.", source="reflection")

---

Core API

All methods are async-first with sync wrappers (add_sync, search_sync).

Method	Description
add()	Store a memory
search()	Hybrid search (lexical + vector + importance + decay)
get()	Retrieve by ID (with access control)
update()	Patch memory fields
delete()	Delete by filter expressions (GDPR bulk delete)
explain()	Retrieval trace with scores, decay factors, matched terms
timeline()	Ordered history of all memory operations
ingest_event()	Process a signal through the policy engine
promote()	Promote episodic patterns to procedural memory
apply_decay()	Apply retention decay to all memories
forget_user()	Right-to-be-forgotten workflow
run_with_attribution()	Execute task and track which memories influenced the output

---

Policy Configuration

Lifecycle policies are declarative YAML — no imperative code required.

# engramx.yaml
driver:
  kind: sqlite
  path: .engramx/engramx.db

policies:
  extraction:
    - name: extract-user-preferences
      trigger: conversation_turn
      conditions:
        - content_matches: [prefer, always, never, like]
      create:
        type: semantic
        scope: user
        importance_score: 0.8
        sensitivity_level: internal

  retention:
    - name: session-episode-decay
      applies_to: { type: episodic, scope: session }
      decay_function: exponential
      half_life_days: 7
      floor_score: 0.1

  summarization:
    - name: promote-repeated-success
      applies_to: { type: episodic }
      trigger: "same_action_success_count >= 3"
      promote_to:
        type: procedural
        scope: agent
        content: "Confirm prerequisites before executing repeated workflows."

  governance:
    - name: gdpr-user-data
      applies_to:
        scope: user
        sensitivity_level: [confidential, restricted]
      retention_days: 30
      on_user_deletion: delete_all
      audit_log: true

from engramx import MemoryClient
client = MemoryClient(config="engramx.yaml")

---

Embedding Providers

EngramX ships with a dependency-free hash embedder and supports real embedding models. The OpenAI embedder uses the OpenAI SDK format which works with any compatible provider via base_url.

from engramx import create_embedder

# Default: hash-based (no dependencies, deterministic)
embedder = create_embedder("hash")

# OpenAI
embedder = create_embedder("openai", model="text-embedding-3-small")

# Google Gemini (OpenAI-compatible endpoint)
embedder = create_embedder("openai",
    model="gemini-embedding-001",
    api_key="GEMINI_KEY",
    base_url="https://generativelanguage.googleapis.com/v1beta/openai/",
)

# Cohere (OpenAI-compatible endpoint)
embedder = create_embedder("openai",
    model="embed-english-v3.0",
    api_key="COHERE_KEY",
    base_url="https://api.cohere.ai/compatibility/v1",
)

# Mistral
embedder = create_embedder("openai",
    model="mistral-embed",
    api_key="MISTRAL_KEY",
    base_url="https://api.mistral.ai/v1",
)

# Together AI
embedder = create_embedder("openai",
    model="togethercomputer/m2-bert-80M-8k-retrieval",
    api_key="TOGETHER_KEY",
    base_url="https://api.together.xyz/v1",
)

# LiteLLM — unified interface for 100+ providers
embedder = create_embedder("litellm", model="cohere/embed-english-v3.0")
embedder = create_embedder("litellm", model="gemini/gemini-embedding-001")
embedder = create_embedder("litellm", model="bedrock/amazon.titan-embed-text-v1")

# Sentence Transformers (local models)
embedder = create_embedder("sentence_transformers", model_name="all-MiniLM-L6-v2")

# Pass to client
client = MemoryClient(driver="memory", embedder=embedder)

Provider	Via	Install
OpenAI	create_embedder("openai")	pip install engramx[openai]
Google Gemini	create_embedder("openai", base_url=...)	pip install engramx[openai]
Cohere	create_embedder("openai", base_url=...)	pip install engramx[openai]
Mistral	create_embedder("openai", base_url=...)	pip install engramx[openai]
Together AI	create_embedder("openai", base_url=...)	pip install engramx[openai]
LiteLLM (100+)	create_embedder("litellm")	pip install engramx[litellm]
Sentence Transformers	create_embedder("sentence_transformers")	pip install engramx[sentence-transformers]
Hash (local, no deps)	create_embedder("hash")	—

---

LLM Reflection

Summarize episodic memory into durable knowledge using LLM-powered reflection:

from engramx import create_reflector

# OpenAI
reflector = create_reflector("openai", model="gpt-4o-mini")

# Any OpenAI-compatible provider via base_url
reflector = create_reflector("openai",
    model="gemini-2.0-flash",
    api_key="GEMINI_KEY",
    base_url="https://generativelanguage.googleapis.com/v1beta/openai/",
)

# Anthropic
reflector = create_reflector("anthropic", model="claude-sonnet-4-20250514")

# LiteLLM — any of 100+ providers
reflector = create_reflector("litellm", model="anthropic/claude-sonnet-4-20250514")
reflector = create_reflector("litellm", model="gemini/gemini-2.0-flash")
reflector = create_reflector("litellm", model="cohere/command-r-plus")

---

Framework Adapters

LangChain

from engramx import MemoryClient
from engramx.adapters import EngramChatMemory

client = MemoryClient(driver="sqlite")
memory = EngramChatMemory(client=client, user_id="u-123")

# Works as a LangChain memory drop-in
context = memory.load_memory_variables({"input": "What units do I prefer?"})
memory.save_context({"input": "Use metric"}, {"output": "Noted!"})

LlamaIndex

from engramx import MemoryClient
from engramx.adapters import EngramMemoryBlock

client = MemoryClient(driver="sqlite")
block = EngramMemoryBlock(client=client, user_id="u-123")

context = block.get("unit preferences")
block.put("User prefers metric units.")

AutoGen

from engramx import MemoryClient
from engramx.adapters import EngramAutoGenMemory

client = MemoryClient(driver="sqlite")
memory = EngramAutoGenMemory(client=client, user_id="u-123")

results = memory.query("payment preferences")
memory.add("User prefers credit card payments.")

LangGraph

Use EngramX as long-term memory in LangGraph agents. EngramX handles cross-session memory while LangGraph checkpoints handle short-term thread state.

from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode
from langchain_openai import ChatOpenAI
from typing import Annotated, TypedDict
import operator

from engramx import MemoryClient

# Initialize EngramX for long-term memory
engramx = MemoryClient(driver="sqlite")
model = ChatOpenAI(model="gpt-4o-mini")


# 1. Define graph state
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    memory_context: str
    user_id: str


# 2. Memory retrieval node — runs before LLM
async def recall_memories(state: AgentState) -> dict:
    """Retrieve relevant EngramX memories and inject into context."""
    user_msg = state["messages"][-1].content if state["messages"] else ""
    user_id = state.get("user_id", "default")

    memories = await engramx.search(
        query=user_msg,
        filters={"user_id": user_id},
        top_k=5,
    )

    if memories:
        context = "\n".join(f"- {m.record.content}" for m in memories)
    else:
        context = "No relevant memories found."

    return {"memory_context": context}


# 3. LLM node — uses memories as context
async def call_llm(state: AgentState) -> dict:
    """Call LLM with memory-augmented context."""
    system_msg = {
        "role": "system",
        "content": (
            "You are a helpful assistant with memory.\n"
            f"Relevant memories:\n{state.get('memory_context', 'None')}"
        ),
    }
    response = await model.ainvoke([system_msg] + state["messages"])
    return {"messages": [response]}


# 4. Memory persistence node — runs after LLM response
async def persist_memories(state: AgentState) -> dict:
    """Store the conversation turn as episodic memory in EngramX."""
    user_id = state.get("user_id", "default")
    messages = state["messages"]

    if len(messages) >= 2:
        user_msg = messages[-2].content
        ai_msg = messages[-1].content
        await engramx.add(
            type="episodic",
            scope="user",
            user_id=user_id,
            content=f"User: {user_msg}\nAssistant: {ai_msg}",
            source="conversation",
            importance_score=0.7,
        )

    return {}


# 5. Build the graph
graph = StateGraph(AgentState)
graph.add_node("recall", recall_memories)
graph.add_node("llm", call_llm)
graph.add_node("persist", persist_memories)

graph.add_edge(START, "recall")
graph.add_edge("recall", "llm")
graph.add_edge("llm", "persist")
graph.add_edge("persist", END)

agent = graph.compile()

# 6. Run
# result = await agent.ainvoke({
#     "messages": [HumanMessage(content="What units should I use?")],
#     "user_id": "u-123",
# })

LangGraph + EngramX architecture:

┌─────────────────────────────────────────────────┐
│                  LangGraph Agent                │
│                                                 │
│  START → [recall] → [llm] → [persist] → END    │
│             │                    │               │
│             ▼                    ▼               │
│     ┌──────────────┐    ┌──────────────┐        │
│     │    EngramX   │    │    EngramX   │        │
│     │   search()    │    │    add()      │        │
│     │  (long-term)  │    │  (long-term)  │        │
│     └──────────────┘    └──────────────┘        │
│                                                 │
│  Short-term state: LangGraph checkpoints        │
│  Long-term memory: EngramX (any backend)         │
└─────────────────────────────────────────────────┘

---

Agent with Attribution

Track which memories influenced agent output:

from engramx import MemoryClient

client = MemoryClient(driver="memory")

async def my_llm(prompt, memories):
    return "Response based on memories"

response, attribution = await client.run_with_attribution(
    "Book a flight to Tokyo",
    user_id="u-123",
    runner=my_llm,
)

print(attribution.memories_used)    # ["mem-abc", "mem-def"]
print(attribution.policies_fired)   # ["extract-user-preferences"]

Or use the standalone EngramAgent with OpenAI:

from engramx import MemoryClient, EngramAgent

client = MemoryClient(driver="sqlite")
agent = EngramAgent(client=client, model="gpt-4o-mini")

response, attribution = await agent.run_with_attribution(
    "What units should I use?",
    user_id="u-123",
)

---

Observability

Every retrieval is traceable:

trace = await client.explain(query="unit preferences", filters={"user_id": "u-123"})
print(trace.to_markdown())

Output:

# Memory Trace
- Query: `unit preferences`
- Filters: `{'user_id': 'u-123'}`

| Memory ID | Score | Decay Adjusted | Matched Terms | Policy Filters |
| --- | ---: | ---: | --- | --- |
| `mem-abc-123` | 0.870 | 0.740 | metric, unit | scope=user |

Timeline of all operations:

events = await client.timeline(user_id="u-123", types=["episodic", "semantic"])
for event in events:
    print(f"{event.timestamp} {event.operation} {event.memory_id}")

---

Background Jobs

Schedule automatic decay, promotion, and governance enforcement:

from engramx import MemoryClient, PolicyJobScheduler

client = MemoryClient(config="engramx.yaml")
scheduler = PolicyJobScheduler(
    driver=client.driver,
    policy_engine=client.policy_engine,
    interval_seconds=300,  # Run every 5 minutes
)

# Run once
report = await scheduler.run_once()
print(f"Decayed: {len(report.decay_updated_ids)}")
print(f"Promoted: {len(report.promoted_ids)}")
print(f"Deleted: {len(report.deleted_ids)}")

# Or run continuously
await scheduler.start()
# ... later ...
await scheduler.stop()

---

Chat Example

Example chat loop with persistent memory:

from openai import OpenAI
from engramx import MemoryClient

openai_client = OpenAI()
memory = MemoryClient(driver="sqlite")

def chat_with_memories(message: str, user_id: str = "default_user") -> str:
    relevant = memory.search_sync(query=message, filters={"user_id": user_id}, top_k=3)
    memories_str = "\n".join(f"- {m.record.content}" for m in relevant) or "- None yet."

    response = openai_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": f"You are a helpful AI.\nUser Memories:\n{memories_str}"},
            {"role": "user", "content": message},
        ],
    )
    reply = response.choices[0].message.content or ""

    memory.add_sync(
        type="episodic", scope="user", user_id=user_id,
        content=f"User: {message}\nAssistant: {reply}",
        source="conversation", importance_score=0.7,
    )
    return reply

while True:
    user_input = input("You: ").strip()
    if user_input.lower() == "exit":
        break
    print(f"AI: {chat_with_memories(user_input)}")

---

Testing

pip install -e .[dev]
python -m pytest -q

The test suite covers storage drivers, the policy engine, test harness, adapters, background jobs, retrieval features, and optional driver dependency checks.

---

Development

git clone https://github.com/TechyNilesh/EngramX.git
cd EngramX
python -m venv .venv
source .venv/bin/activate
pip install -e .[dev]
python -m pytest -q

---

Architecture

engramx/
  schema.py              # MemoryRecord, MemoryType, MemoryScope, SensitivityLevel
  client.py              # MemoryClient — unified API
  config.py              # YAML config loading
  policy.py              # PolicyEngine — extraction, decay, promotion, governance
  lifecycle.py           # Lifecycle functions — extract, decay, summarize, governance
  observability.py       # MemoryTrace, ScoredMemory, OutputAttribution, MemoryTimelineEvent
  embedding.py           # Hash, OpenAI, SentenceTransformer embedders
  reflection.py          # LLM reflectors (OpenAI, Anthropic)
  agent.py               # EngramAgent with run_with_attribution()
  jobs.py                # PolicyJobScheduler for background lifecycle jobs
  models.py              # MemorySignal (event model)
  testing.py             # MemoryHarness for CI
  storage/
    base.py              # BaseDriver abstract interface
    memory.py            # InMemoryDriver
    sqlite.py            # SQLiteDriver
    postgres.py          # PostgresDriver (asyncpg + pgvector)
    chroma.py            # ChromaDriver
    qdrant.py            # QdrantDriver
    redis.py             # RedisDriver
    neo4j.py             # Neo4jDriver
    mem0.py              # Mem0Driver (delegation)
    zep.py               # ZepDriver (delegation)
    ranking.py           # Hybrid scoring (lexical + vector + importance + decay)
  adapters/
    langchain.py         # EngramChatMemory
    llamaindex.py        # EngramMemoryBlock
    autogen.py           # EngramAutoGenMemory
    chat.py              # EngramChatAdapter

---

Core Contributor

Nilesh Verma

---

Citation

@software{verma2026engramx,
  author = {Nilesh Verma},
  title = {EngramX: The Open-Source Agent Memory Framework},
  year = {2026},
  url = {https://github.com/TechyNilesh/EngramX},
  version = {0.1.0}
}

---

License

MIT