BaseMemory & BaseRetriever

Purpose

Two abstract base classes define the extension points of this package: BaseMemory for conversation-memory-style components (save/load/clear), and BaseRetriever for retrieval-style components (add documents / get relevant documents). They are unrelated class hierarchies — neither inherits from the other.

config = MemoryConfig()   # module-level singleton, unused by either class below

[!NOTE] Module-level config is unused base.py instantiates a module-level MemoryConfig() on import but neither BaseMemory nor BaseRetriever reference it anywhere in the module. It appears to be dead code / a leftover from an earlier version.

BaseMemory

class BaseMemory(ABC):
    @abstractmethod
    def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, Any]): ...

    @abstractmethod
    def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]: ...

    @abstractmethod
    def clear(self): ...

    def get_memory_stats(self) -> Dict[str, Any]:
        return {'type': self.__class__.__name__, 'size': 0}

Abstract methods

Concrete method

get_memory_stats() has a default implementation returning {'type': ClassName, 'size': 0}. Every concrete subclass in this package overrides this with a richer, class-specific stats dict — the base implementation's hardcoded 'size': 0 is never actually returned in practice by any of the four built-in memories.

[!WARNING] No async methods on BaseMemory Unlike BaseRetriever (below), BaseMemory does not declare asave_context / aload_memory_variables. If you implement a custom BaseMemory subclass and want async support consistent with the built-in classes, you must add these two methods yourself — see Architecture → Concurrency Model and Known Issues.

Implementations

All four classes in memory.memories implement BaseMemory: ConversationBufferMemory, ConversationBufferWindowMemory, ConversationSummaryMemory, ConversationEntityMemory.

BaseRetriever

class BaseRetriever(ABC):
    @abstractmethod
    def add_documents(self, documents: List[Any]): ...

    @abstractmethod
    def get_relevant_documents(self, query: str, k: int = 3) -> List[Any]: ...

    async def asave_context(self, inputs, outputs):
        import asyncio
        return await asyncio.to_thread(self.save_context, inputs, outputs)

    async def aload_memory_variables(self, inputs):
        import asyncio
        return await asyncio.to_thread(self.load_memory_variables, inputs)

    async def __aenter__(self):
        return self

    async def __aexit__(self, *_):
        self.clear()
        return False

Abstract methods

[!DANGER] asave_context / aload_memory_variables / clear are defined but reference undeclared methods BaseRetriever.asave_context calls self.save_context, and __aexit__ calls self.clear() — but BaseRetriever itself declares neither save_context nor clear (those belong to the unrelated BaseMemory interface). Any concrete BaseRetriever subclass that doesn't separately define save_context and clear will raise AttributeError the moment asave_context or the async with retriever: context-manager exit path is used. No class in this source tree actually subclasses BaseRetriever (grep across the package finds zero references), so this defect has not manifested in any shipped code path — but it is a latent trap for anyone implementing a custom retriever against this base class as documented. See Known Issues.

Async context manager

BaseRetriever implements __aenter__/__aexit__, making any subclass usable as async with MyRetriever() as r: ... — but as noted above, __aexit__ will fail unless the subclass also defines clear().

Implementations

No class in this source tree subclasses BaseRetriever. It is exported (from zerda.memory import BaseRetriever) for external use but has no built-in implementation to reference as an example.

Thread safety

Neither class holds any state of its own (aside from the unused module-level config), so thread safety is entirely a property of each concrete subclass.

Examples

from zerda.memory.core import BaseMemory
from typing import Dict, Any

class MyCustomMemory(BaseMemory):
    def __init__(self):
        self.data = []

    def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, Any]):
        self.data.append((inputs, outputs))

    def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
        return {"history": self.data}

    def clear(self):
        self.data.clear()

    # Add these yourself for async support — BaseMemory doesn't provide them:
    async def asave_context(self, inputs, outputs):
        import asyncio
        return await asyncio.to_thread(self.save_context, inputs, outputs)

    async def aload_memory_variables(self, inputs):
        import asyncio
        return await asyncio.to_thread(self.load_memory_variables, inputs)

Best practices

• When implementing BaseMemory, follow the pattern used by the four built-in classes: define your own asave_context/aload_memory_variables locally rather than assuming inheritance provides them.
• When implementing BaseRetriever, also define save_context and clear even though they aren't declared abstract on this class — otherwise the inherited async/context-manager methods will crash.

Related APIs

• Conversation Memories — all four implement BaseMemory.
• Architecture → Extension Points

Version history

Not determinable from this source tree.