Custom Memory Adapters

Plug in your own memory backend with the adapter registry

Next

from praisonaiagents import Agent, MemoryConfig

agent = Agent(
    name="memory-agent",
    instructions="Remember facts across turns.",
    memory=MemoryConfig(provider="my_backend"),
)
agent.start("Store that my favourite colour is teal.")

PraisonAI’s memory backends are pluggable — register your own adapter to store agent memory in any system.

The user chats with the agent; your custom memory adapter persists and retrieves facts from your chosen backend.

Quick Start

1. Simple Usage

   Use a built-in adapter via string to anchor the mental model:

   from praisonaiagents import Agent
   
   agent = Agent(
       name="assistant",
       instructions="Remember user preferences",
       memory="in_memory",  # Ephemeral dict-backed storage
   )
   agent.start("My favourite colour is blue")

2. With a Custom Adapter

   Register your adapter, then point the agent at it:

   from typing import Any, Dict, List, Optional
   from praisonaiagents import Agent, add_memory_adapter
   
   
   class RedisMemoryAdapter:
       def __init__(self, **kwargs):
           self._store: List[Dict[str, Any]] = []
   
       def store_short_term(self, text: str, metadata: Optional[Dict[str, Any]] = None, **kwargs) -> str:
           doc_id = str(len(self._store))
           self._store.append({"id": doc_id, "text": text, "metadata": metadata or {}})
           return doc_id
   
       def search_short_term(
           self,
           query: str,
           limit: int = 5,
           metadata_filter: Optional[Dict[str, Any]] = None,
           user_id: Optional[str] = None,
           **kwargs,
       ) -> List[Dict[str, Any]]:
           hits = [e for e in self._store if query.lower() in e["text"].lower()]
           if user_id is not None:
               hits = [e for e in hits if e["metadata"].get("user_id") == user_id]
           if metadata_filter:
               hits = [e for e in hits if all(e["metadata"].get(k) == v for k, v in metadata_filter.items())]
           return hits[:limit]
   
       def store_long_term(self, text: str, metadata: Optional[Dict[str, Any]] = None, **kwargs) -> str:
           return self.store_short_term(text, metadata, **kwargs)
   
       def search_long_term(
           self,
           query: str,
           limit: int = 5,
           metadata_filter: Optional[Dict[str, Any]] = None,
           user_id: Optional[str] = None,
           **kwargs,
       ) -> List[Dict[str, Any]]:
           return self.search_short_term(query, limit, metadata_filter=metadata_filter, user_id=user_id, **kwargs)
   
       def get_all_memories(self, **kwargs) -> List[Dict[str, Any]]:
           return list(self._store)
   
   
   add_memory_adapter("redis", RedisMemoryAdapter)
   
   agent = Agent(
       name="assistant",
       memory={"provider": "redis"},
   )

The Adapter Is the Single Source of Truth

Previously, search, reset, and delete only delegated to the adapter when the provider was hardcoded sqlite or in_memory. Any other registered adapter stored data correctly but reads hit the empty legacy tables and returned [] — memory was effectively write-only. All four operation paths now gate symmetrically on whether an adapter is configured.

Round-trip through a custom adapter — store then find:

from praisonaiagents import Agent

agent = Agent(
    name="Assistant",
    instructions="Remember what the user tells you.",
    memory={"provider": "dakera"},  # or any registered adapter
)

agent.start("Remember that my project deadline is Friday.")
agent.start("What did I tell you about my deadline?")  # now finds it

How It Works

When Memory initialises, it resolves the provider through the adapter registry — the only code path for backend setup since PR #2060 removed orphaned legacy _init_* methods.

Every operation routes through the adapter

Once you register an adapter (e.g. add_memory_adapter("dakera", DakeraAdapter)) and set provider="dakera" — or pass the adapter directly — every read and every write routes through it: store_, search_, reset_, and delete_. There is no provider in {sqlite, in_memory} gate.

Before this became symmetric, a registered adapter’s writes went to its own table while every search_* looked at the legacy long_mem table the adapter never created — so searches returned [] and resets/deletes no-op’d. Store, search, reset, and delete now share one source of truth. The dedicated chroma / mem0 / mongodb write paths still take priority for those providers.

Built-in Adapters

Heavy backends register as factories in praisonaiagents.memory.adapters.factories so optional dependencies load only when requested.

Adapter operations (as of PR #3515): store ✅ · search ✅ · reset ✅ · delete ✅ — every operation routes through the configured adapter, for any provider, not just sqlite / in_memory.

Reset Capability

Memory.reset_short_term() / .reset_long_term() delegate to the active adapter. Adapters that don’t implement the method log a warning stating the tier was NOT cleared.

Register Your Own Adapter

Implement MemoryProtocol — at minimum: store_short_term, search_short_term, store_long_term, search_long_term, and get_all_memories.

from praisonaiagents import add_memory_adapter, get_memory_adapter

add_memory_adapter("my_backend", MyAdapter)
adapter = get_memory_adapter("my_backend")

Then use it from an agent:

agent = Agent(memory={"provider": "my_backend"})

Registry API at a Glance

Common Patterns

Async adapter — implement AsyncMemoryProtocol (astore_short_term, asearch_short_term, etc.) when your backend is async-native.

Lazy-loaded heavy backend — use add_memory_factory(name, create_fn) (or its synonym register_memory_factory) so imports like chromadb or pymongo happen inside the factory, not at package import time.

Extending an existing adapter — subclass SqliteMemoryAdapter or wrap InMemoryAdapter and register under a new name.

Inspect the registry — check what’s registered before wiring an agent:

from praisonaiagents import has_memory_adapter, list_memory_adapters, add_memory_adapter

if not has_memory_adapter("redis"):
    add_memory_adapter("redis", RedisMemoryAdapter)

print(list_memory_adapters())  # ['sqlite', 'in_memory', 'mem0', 'chroma', 'mongodb', 'dakera', 'redis']

Best Practices

from praisonaiagents import list_memory_adapters
print(list_memory_adapters())  # ['sqlite', 'in_memory', 'mem0', 'chroma', 'mongodb', 'dakera', ...]

import threading

class MyAdapter:
    def __init__(self):
        self._lock = threading.RLock()
        self._store = {}

    def store_long_term(self, memory_id, content, **kwargs):
        with self._lock:
            self._store[memory_id] = content

Related